SingleOutputStreamOperator 写入hive样例

以下是将 Flink 的 SingleOutputStreamOperator 写入 Hive 的样例代码：

// 创建 StreamExecutionEnvironment
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 创建 StreamTableEnvironment
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

// 读取 Kafka 数据流
DataStream<String> dataStream = env.addSource(new FlinkKafkaConsumer<>("topic", new SimpleStringSchema(), properties));

// 将数据流转换为 Table 对象
Table table = tableEnv.fromDataStream(dataStream, "field1, field2, field3");

// 将 Table 注册为临时表
tableEnv.createTemporaryView("temp_table", table);

// 定义 Hive 表的 schema
String schema = "CREATE TABLE hive_table (field1 STRING, field2 BIGINT, field3 DECIMAL(10, 2)) STORED AS ORC";

// 在 Hive 中创建表
tableEnv.executeSql(schema);

// 将临时表数据写入 Hive 表
tableEnv.executeSql("INSERT INTO hive_table SELECT field1, field2, field3 FROM temp_table");

需要注意的是，你需要在 Flink 和 Hive 之间配置正确的连接和认证信息，以及正确的表路径和表名。默认情况下，Flink 的数据会写入到 Hive 的 default 数据库中。