modify source name
This commit is contained in:
parent
75854c56a7
commit
f9ce37ad80
|
@ -263,7 +263,7 @@ splitSql.setSelect("ts, current, voltage, phase, groupid, location")
|
|||
Class<SourceRecords<RowData>> typeClass = (Class<SourceRecords<RowData>>) (Class<?>) SourceRecords.class;
|
||||
SourceSplitSql sql = new SourceSplitSql("select ts, `current`, voltage, phase, tbname from meters");
|
||||
TDengineSource<SourceRecords<RowData>> source = new TDengineSource<>(connProps, sql, typeClass);
|
||||
DataStreamSource<SourceRecords<RowData>> input = env.fromSource(source, WatermarkStrategy.noWatermarks(), "kafka-source");
|
||||
DataStreamSource<SourceRecords<RowData>> input = env.fromSource(source, WatermarkStrategy.noWatermarks(), "tdengine-source");
|
||||
DataStream<String> resultStream = input.map((MapFunction<SourceRecords<RowData>, String>) records -> {
|
||||
StringBuilder sb = new StringBuilder();
|
||||
Iterator<RowData> iterator = records.iterator();
|
||||
|
@ -304,7 +304,7 @@ splitSql.setSelect("ts, current, voltage, phase, groupid, location")
|
|||
config.setProperty(TDengineCdcParams.VALUE_DESERIALIZER, "RowData");
|
||||
config.setProperty(TDengineCdcParams.VALUE_DESERIALIZER_ENCODING, "UTF-8");
|
||||
TDengineCdcSource<RowData> tdengineSource = new TDengineCdcSource<>("topic_meters", config, RowData.class);
|
||||
DataStreamSource<RowData> input = env.fromSource(tdengineSource, WatermarkStrategy.noWatermarks(), "kafka-source");
|
||||
DataStreamSource<RowData> input = env.fromSource(tdengineSource, WatermarkStrategy.noWatermarks(), "tdengine-source");
|
||||
DataStream<String> resultStream = input.map((MapFunction<RowData, String>) rowData -> {
|
||||
StringBuilder sb = new StringBuilder();
|
||||
sb.append("tsxx: " + rowData.getTimestamp(0, 0) +
|
||||
|
@ -343,7 +343,7 @@ splitSql.setSelect("ts, current, voltage, phase, groupid, location")
|
|||
|
||||
Class<ConsumerRecords<RowData>> typeClass = (Class<ConsumerRecords<RowData>>) (Class<?>) ConsumerRecords.class;
|
||||
TDengineCdcSource<ConsumerRecords<RowData>> tdengineSource = new TDengineCdcSource<>("topic_meters", config, typeClass);
|
||||
DataStreamSource<ConsumerRecords<RowData>> input = env.fromSource(tdengineSource, WatermarkStrategy.noWatermarks(), "kafka-source");
|
||||
DataStreamSource<ConsumerRecords<RowData>> input = env.fromSource(tdengineSource, WatermarkStrategy.noWatermarks(), "tdengine-source");
|
||||
DataStream<String> resultStream = input.map((MapFunction<ConsumerRecords<RowData>, String>) records -> {
|
||||
Iterator<ConsumerRecord<RowData>> iterator = records.iterator();
|
||||
StringBuilder sb = new StringBuilder();
|
||||
|
@ -388,7 +388,7 @@ splitSql.setSelect("ts, current, voltage, phase, groupid, location")
|
|||
config.setProperty(TDengineCdcParams.VALUE_DESERIALIZER, "com.taosdata.flink.entity.ResultDeserializer");
|
||||
config.setProperty(TDengineCdcParams.VALUE_DESERIALIZER_ENCODING, "UTF-8");
|
||||
TDengineCdcSource<ResultBean> tdengineSource = new TDengineCdcSource<>("topic_meters", config, ResultBean.class);
|
||||
DataStreamSource<ResultBean> input = env.fromSource(tdengineSource, WatermarkStrategy.noWatermarks(), "kafka-source");
|
||||
DataStreamSource<ResultBean> input = env.fromSource(tdengineSource, WatermarkStrategy.noWatermarks(), "tdengine-source");
|
||||
DataStream<String> resultStream = input.map((MapFunction<ResultBean, String>) rowData -> {
|
||||
StringBuilder sb = new StringBuilder();
|
||||
sb.append("ts: " + rowData.getTs() +
|
||||
|
|
Loading…
Reference in New Issue