Dear All,
My name is Han. I'm very interested in your advanced Flink system, and I'm
learning it.
I'm writing to your group for communicating about my personal question. I
tried to use Table API&SQL and register a TableSource by the
KafkaJsonTableSource method, I have to say it works very well. My code as
follows.
And my question is about TypeInformation. The first parameter returns an array
of String types and it contains fields in the kafka, the second parameter
corresponds to the type of each field,i.e the below code,new
String[]{"a","b","c"} is the the first parameter,new
TypeInformation<?>[]{Types.STRING(),Types.STRING(),Types.STRING()} is the
second parameter.
If I didn't know the name of the field in the kafka before registering
TableSource, in other words,
the fields in the kafka are dynamic, how to solve this problem?
TypeInformation<Row> typeInfo = Types.ROW(
new String[]{"a","b","c"},
new TypeInformation<?>[]{Types.STRING(),Types.STRING(),Types.STRING()}
);
KafkaJsonTableSource kafkaTableSource = new
Kafka010JsonTableSource(eltResultTopic,
inputProperties, typeInfo);
kafkaTableSource.setFailOnMissingField(false);
tableEnvironment.registerTableSource("kafkaSource",kafkaTableSource);
Table sqlResult = tableEnvironment.sql("select a from kafkaSource");
DataStream<String> dataStream =
tableEnvironment.toAppendStream(sqlResult,String.class);
dataStream.print();
environment.execute();
Waiting for your earlier reply. Thanks.
Best Wishes,
Han
