flink table api 在同一个job 怎么输出到多个表 ,设置jobid,命名该job 的名称。使用java语言
时间: 2024-06-11 20:05:45 浏览: 7
在 Flink Table API 中,可以使用 `insertInto()` 方法将数据流输出到一个表中。要输出到多个表中,可以使用 Flink 提供的 `TableSink` 接口,自定义一个 `TableSink` 实现类,然后在 Table API 中使用这个自定义的 `TableSink`。具体步骤如下:
1. 自定义一个 `TableSink` 实现类,可以参考 Flink 官方提供的一些实现类,如 `CsvTableSink`、`JdbcTableSink` 等。
```java
public class MyTableSink implements TableSink<Row> {
private String[] fieldNames;
private TypeInformation[] fieldTypes;
private String tableName;
public MyTableSink(String[] fieldNames, TypeInformation[] fieldTypes, String tableName) {
this.fieldNames = fieldNames;
this.fieldTypes = fieldTypes;
this.tableName = tableName;
}
@Override
public DataType getConsumedDataType() {
return DataTypes.createRowType(fieldTypes, fieldNames);
}
@Override
public TableSink<Row> configure(String[] fieldNames, TypeInformation[] fieldTypes) {
return new MyTableSink(fieldNames, fieldTypes, tableName);
}
@Override
public void emitDataStream(DataStream<Row> dataStream) {
dataStream.addSink(new MySinkFunction(tableName)).name(tableName);
}
@Override
public TypeInformation<Row> getOutputType() {
return Types.ROW_NAMED(fieldNames, fieldTypes);
}
@Override
public String[] getFieldNames() {
return fieldNames;
}
@Override
public TypeInformation<?>[] getFieldTypes() {
return fieldTypes;
}
}
```
2. 在 Table API 中使用自定义的 `TableSink`,将数据流输出到多个表中。需要为每个 `Table` 设置一个不同的 `tableName`,并使用 `insertInto()` 方法将数据流插入到对应的表中。
```java
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
// 创建表
Table table1 = tEnv.fromDataStream(dataStream1, "field1, field2");
Table table2 = tEnv.fromDataStream(dataStream2, "field1, field3");
// 输出到多个表中
table1.insertInto(new MyTableSink(new String[]{"field1", "field2"}, new TypeInformation<?>[]{Types.STRING, Types.INT}, "table1"));
table2.insertInto(new MyTableSink(new String[]{"field1", "field3"}, new TypeInformation<?>[]{Types.STRING, Types.DOUBLE}, "table2"));
env.execute("jobName");
```
3. 可以设置 `jobId` 和 `jobName`,方式如下:
```java
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
env.enableCheckpointing(10000);
env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
env.getCheckpointConfig().setCheckpointTimeout(60000);
env.getCheckpointConfig().setMaxConcurrentCheckpoints(1);
env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 5000));
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
// 设置 jobId 和 jobName
env.getStreamGraph().setJobName("MyJob");
env.getStreamGraph().setJobId("jobId_001");
// 创建表
Table table1 = tEnv.fromDataStream(dataStream1, "field1, field2");
Table table2 = tEnv.fromDataStream(dataStream2, "field1, field3");
// 输出到多个表中
table1.insertInto(new MyTableSink(new String[]{"field1", "field2"}, new TypeInformation<?>[]{Types.STRING, Types.INT}, "table1"));
table2.insertInto(new MyTableSink(new String[]{"field1", "field3"}, new TypeInformation<?>[]{Types.STRING, Types.DOUBLE}, "table2"));
env.execute("MyJob");
```
相关推荐
![zip](https://img-home.csdnimg.cn/images/20210720083736.png)
![zip](https://img-home.csdnimg.cn/images/20210720083736.png)
![zip](https://img-home.csdnimg.cn/images/20210720083736.png)
![](https://csdnimg.cn/download_wenku/file_type_ask_c1.png)
![](https://csdnimg.cn/download_wenku/file_type_ask_c1.png)
![](https://csdnimg.cn/download_wenku/file_type_ask_c1.png)
![](https://csdnimg.cn/download_wenku/file_type_ask_c1.png)
![](https://csdnimg.cn/download_wenku/file_type_ask_c1.png)
![](https://csdnimg.cn/download_wenku/file_type_ask_c1.png)
![](https://csdnimg.cn/download_wenku/file_type_ask_c1.png)
![](https://csdnimg.cn/download_wenku/file_type_ask_c1.png)
![](https://csdnimg.cn/download_wenku/file_type_ask_c1.png)
![](https://csdnimg.cn/download_wenku/file_type_ask_c1.png)
![](https://csdnimg.cn/download_wenku/file_type_ask_c1.png)
![](https://csdnimg.cn/download_wenku/file_type_ask_c1.png)
![](https://csdnimg.cn/download_wenku/file_type_ask_c1.png)
![](https://csdnimg.cn/download_wenku/file_type_ask_c1.png)
![](https://csdnimg.cn/download_wenku/file_type_ask_c1.png)