批量日志数据库外表写入

#创建外链表映射日志文件

audience_attributes_path = "gphdfs://xxx/audience_attributes/#{batch_id}"

create_sql = <<-eos
        create writable external table #{audience_attributes_table} (
          opxpid varchar(60),
          hash_key varchar(50),
          hash_value text
        )
       
        location ('#{audience_attributes_path}')
        format 'text';
        eos

AudienceStructure.connection.execute(create_sql)

#表数据插入

insert_sql = <<-sql
        insert into #{audience_attributes_table}
        select opxpid, hash_key, array_to_string(jx_array_sort(madlib.array_agg(audience_id)), ',')
        from (
        select opxpid, hash_key, hash_value audience_id
        from #{master_table}
        where hash_key = 'aids'
        and hash_value in (#{aids.uniq.join(",")})
        ) m
        group by 1, 2;
        sql

AudienceStructure.connection.execute(insert_sql)


#保留12个表超过12的删除。

drop_tables("select 'drop external table workspace.' || relname || ';' cmd from pg_class where relname like 'audience_segments_hdfs_%' order by 1")

def drop_tables(sql, col_name="cmd")
        drop_sqls = AudienceStructure.connection.select_all(sql)
        # only keep latest 12 tables
        if drop_sqls.size > 12
          drop_sqls[0..drop_sqls.size-13].each do |drop_sql|
            AudienceStructure.connection.execute(drop_sql[col_name])
          end
        end
   end