flink 自定义批量落地hivesink

要将 Flink 中的数据批量写入到 Hive 中，可以使用 Flink 的 `StreamingFileSink` 将数据先写入到本地文件系统中，然后再使用 Flink 的 `HadoopOutputFormat` 将数据批量写入到 Hive 中。

以下是一个示例代码：

// 创建 StreamingFileSink
StreamingFileSink<String> sink = StreamingFileSink
    .forRowFormat(new Path("/path/to/local/files"), new SimpleStringEncoder<String>("UTF-8"))
    .withRollingPolicy(
        DefaultRollingPolicy.builder()
            .withRolloverInterval(TimeUnit.MINUTES.toMillis(15))
            .withInactivityInterval(TimeUnit.MINUTES.toMillis(5))
            .withMaxPartSize(1024 * 1024 * 1024)
            .build())
    .build();

// 将数据写入到 StreamingFileSink 中
DataStream<String> dataStream = ...
dataStream.addSink(sink);

// 创建 HadoopOutputFormat
HiveConf conf = new HiveConf();
HadoopOutputFormat<Text, NullWritable> outputFormat = new HiveOutputFormat<Text, NullWritable>(conf, "myTable");

// 使用 HadoopOutputFormat 将数据批量写入到 Hive 中
Job job = Job.getInstance();
HadoopOutputFormat.setOutput(job, outputFormat);
FileInputFormat.setInputPaths(job, "/path/to/local/files");
outputFormat.getConfiguration().set(TableOutputFormat.OUTPUT_TABLE, "myTable");
outputFormat.getConfiguration().set(TableOutputFormat.OUTPUT_SCHEMA, "col1 STRING, col2 INT");
outputFormat.getConfiguration().set(TableOutputFormat.OUTPUT_CONF_PREFIX + "mapreduce.job.reduces", "1");
outputFormat.getConfiguration().set(TableOutputFormat.OUTPUT_CONF_PREFIX + "mapreduce.job.name", "myJob");
outputFormat.getConfiguration().set(TableOutputFormat.OUTPUT_CONF_PREFIX + "mapreduce.output.fileoutputformat.outputdir", "/tmp/hive-output");
outputFormat.getConfiguration().set(TableOutputFormat.OUTPUT_CONF_PREFIX + "mapreduce.output.fileoutputformat.compress", "false");
outputFormat.getConfiguration().set(TableOutputFormat.OUTPUT_CONF_PREFIX + "mapreduce.output.fileoutputformat.compress.type", "NONE");
outputFormat.getConfiguration().set(TableOutputFormat.OUTPUT_CONF_PREFIX + "mapreduce.output.fileoutputformat.compress.codec", "org.apache.hadoop.io.compress.DefaultCodec");
outputFormat.getConfiguration().set(TableOutputFormat.OUTPUT_CONF_PREFIX + "mapreduce.output.fileoutputformat.compress.codec.null", "org.apache.hadoop.io.compress.DefaultCodec");

job.waitForCompletion(true);

首先，创建一个 `StreamingFileSink`，将数据写入到本地文件系统中。然后，创建一个 `HadoopOutputFormat`，将数据批量写入到 Hive 中。需要注意的是，需要在 `HadoopOutputFormat` 中指定要写入的表名、表结构和输出路径等参数。在示例代码中，输出路径为 `/tmp/hive-output`，可以根据实际情况进行修改。