flink 读取hive表写入到kafka方案

水深无声 2023-01-22 09:56 2阅读 0赞

场景：

使用flink sql读取hive表数据写入到kafka

hive表大约80个字段 500W数据

遇到问题：

内存爆炸，4G内存都不够，这肯定不行。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70][]

GC严重：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 1][]

解决方案：

之前的代码是流式读取hive表写入kafka，有问题，内存爆炸，写入也慢。

如果使用batch方式写入的话，这个问题就解决了:

直接上代码：

public class FlinkReadHiveAndWriteMysql_batch {
    
        private static final String KAFKA_SQL = "CREATE TABLE  kafka_sink (\n" +
                "order_number  string,          \n" +
                "device_type  string \n" +
                ") WITH (\n" +
                " 'connector' = 'kafka',\n" +
                " 'topic' = 'bbb',\n" +
                " 'properties.bootstrap.servers' = 'dev-ct6-dc-worker01:9092,dev-ct6-dc-worker02:9092,dev-ct6-dc-worker03:9092',\n" +
                " 'properties.group.id' = 'test1',\n" +
                " 'format' = 'json',\n" +
                " 'scan.startup.mode' = 'earliest-offset'\n" +
                ")";
    
     
    
     
    
    
    
        private static final String dim_user = "CREATE TABLE `dim_user` (\n" +
                "  `user_code` STRING,\n" +
                "  `user_uid` STRING,\n" +
                "  `user_name` STRING,\n" +
                "  PRIMARY KEY (user_code) NOT ENFORCED\n" +
                ") WITH (\n" +
                "    'connector' = 'jdbc',\n" +
                "    'url' = 'jdbc:mysql://192.168.x.xx:3306/wide_join?useUnicode=true&characterEncoding=UTF-8&serverTimezone=UTC',\n" +
                "    'table-name' = 'dim_user_copy', -- MySQL中的待插入数据的表\n" +
                "    'driver' = 'com.mysql.jdbc.Driver',\n" +
                "    'username' = 'root',\n" +
                "    'password' = '123456',\n" +
                "    'sink.buffer-flush.interval' = '1s'\n" +
                ")";
    
        public static void main(String[] args) throws Exception {
    
            EnvironmentSettings settings = EnvironmentSettings
                    .newInstance()
                    .useBlinkPlanner()
                    .inBatchMode()
                    .build();
            TableEnvironment tableEnv = TableEnvironment.create(settings);
    
            String name = "myhive";      // Catalog名称，定义一个唯一的名称表示
            String defaultDatabase = "doris_test";  // 默认数据库名称
            String hiveConfDir = "G:\\开发的配置文件";  // hive-site.xml路径
            String version = "1.1.0";       // Hive版本号
            HiveCatalog hive = new HiveCatalog(name, defaultDatabase, hiveConfDir, version);
            StatementSet statementSet = tableEnv.createStatementSet();
            tableEnv.registerCatalog(name, hive);
    
            tableEnv.useCatalog(name);
            Table sqlResult = tableEnv.sqlQuery("select * from doris_test.fs_plt_order_flow");
            tableEnv.executeSql("drop table if exists kafka_sink ");
            tableEnv.executeSql(KAFKA_SQL);
    //        tableEnv.executeSql(dim_user);
            statementSet.addInsert("kafka_sink", sqlResult);
            statementSet.execute();
        }
    }

比对数据量：

![20210713170259674.png][]

![20210713170315806.png][]

注意：

我这个表是进行了主键去重的，如果hive数据导入到kafka要求是有序的，这个得要注意一点，要对数据进行排序之后 写入kafka要按自定义规则分区。

上述方案经过验证，发现会报错，后来发现直接table转streaming写入就好了：

public static void main(String[] args) {
    
    
    
            StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
            env.setParallelism(3);
            env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
            EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
            StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, bsSettings);
             Configuration configuration = tEnv.getConfig().getConfiguration();
            configuration.setString("akka.ask.timeout", "360 s");
            configuration.setString("dfs.client.use.datanode.hostname", "true");
    //        bsEnv.enableCheckpointing(5000);
    
    
            tEnv.getConfig().getConfiguration().setBoolean("table.dynamic-table-options.enabled", true);
            String name = "myhive";      // Catalog名称，定义一个唯一的名称表示
            String defaultDatabase = "doris_test";  // 默认数据库名称
            String hiveConfDir = "G:\\汪小剑的文件夹";  // hive-site.xml路径
            String version = "1.1.0";       // Hive版本号
            HiveCatalog hive = new HiveCatalog(name, defaultDatabase, hiveConfDir, version);
            tEnv.registerCatalog(name, hive);
            tEnv.useCatalog(name);
    //        String columns = "order_number,device_type";
            String columns = "id,order_number,stock_code,address_code,address_name,address_type,company_code,company_name,create_time,update_time,contacts_name,contacts_tel,address_memo";
    //        Table table = tEnv.sqlQuery("select order_number,device_type  from doris_test.fs_plt_order_flow");
            Table table = tEnv.sqlQuery("select id,order_number,stock_code,address_code,address_name,address_type,company_code,company_name,create_time,update_time,contacts_name,contacts_tel,address_memo   from doris_test.order_address_test");
            DataStream<Row> jsonObjectDataStream = tEnv.toAppendStream(table, Row.class);
            SingleOutputStreamOperator<String> order_number = jsonObjectDataStream.flatMap(new RichFlatMapFunction<Row, String>() {
    
                @Override
                public void open(Configuration parameters) throws Exception {
                    super.open(parameters);
                }
    
                @Override
                public void flatMap(Row value, Collector<String> out) throws Exception {
                    int arity = value.getArity();
                    JSONObject json = new JSONObject();
                    for (int i = 0; i < arity; i++) {
                        Object field = value.getField(i);
                        String[] split = columns.split(",");
                        json.put(split[i], field);
                    }
                    out.collect(json.toString());
    
                }
            });

但是会遇到报错的问题：

> Failed to connect to /xxx.xxx.xxx:50010 for block, add to deadNodes and continue

hive是有界数据，flink streaming写入kafka 完了之后任务会报错结束，这里只是一种导数的方案，紧紧是内网测试使用，不推荐生产使用。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70]: /images/20221020/24c7daf02b354ba6b19f719109334b53.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzMxODY2Nzkz_size_16_color_FFFFFF_t_70 1]: /images/20221020/acbdbb954c6d461185daa741584094a1.png
[20210713170259674.png]: /images/20221020/6a7b28ac505b47ac9befff4d36514b9a.png
[20210713170315806.png]: /images/20221020/6a33caf103c640bc8df3e093890e05f2.png