Как определить таблицу apache flink с атрибутом времени строки - PullRequest
Новая
       79

Как определить таблицу apache flink с атрибутом времени строки

2 голосов
/ 22 января 2020

У меня json строк, поступающих в качестве моих данных, я хочу создать из них таблицу. 

StreamTableEnvironment fsTableEnv = StreamTableEnvironment.create(streamExecutionEnvironment, fsSettings);
String allEventsTable = "allEventsTable";
        fsTableEnv.connect(new Kafka()
                            .version("0.11")
                            .topic("events")
                            .property("bootstrap.servers", "localhost:9092")
                            .property("group.id", "dummyquery").startFromLatest())
                .withSchema(new Schema()
                    .field("rule_id", Types.INT)
                    .field("sourceAddress", Types.STRING)
                    .field("deviceProduct", Types.STRING)
                    .field("destHost", Types.STRING)
                    .field("extra", Types.STRING)
                    .field("rowtime", Types.SQL_TIMESTAMP)
                        .rowtime(new Rowtime().timestampsFromField("rowtime").watermarksPeriodicBounded(2000))

                )
                .withFormat(new Json().failOnMissingField(false).deriveSchema())
                .inAppendMode()
                .registerTableSource(allEventsTable);

         Table result = fsTableEnv.sqlQuery("select * from allEventsTable where sourceAddress='12345431'");

        DataStream alert = fsTableEnv.toAppendStream(result, Row.class);
        alert.print();

Однако при выполнении задания я получаю ошибку 

Exception in thread "main" org.apache.flink.table.api.ValidationException: Field 'rowtime' could not be resolved by the field mapping.
    at org.apache.flink.table.sources.TableSourceValidation.resolveField(TableSourceValidation.java:245)
    at org.apache.flink.table.sources.TableSourceValidation.lambda$validateTimestampExtractorArguments$6(TableSourceValidation.java:202)
    at java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:193)
    at java.util.Spliterators$ArraySpliterator.forEachRemaining(Spliterators.java:948)
    at java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:481)
    at java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:471)
    at java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:545)
    at java.util.stream.AbstractPipeline.evaluateToArrayNode(AbstractPipeline.java:260)
    at java.util.stream.ReferencePipeline.toArray(ReferencePipeline.java:438)

Ps. Я использую flink 1.9

Мои json данные, которые я помещаю в события kafka topi c, похожи на 

{"rule_id":"", "rowtime":"2020-07-23 13:10:13","sourceAddress":"12345433","deviceProduct":"234r5t", "destHost":"876543", "extra":"dummy"}

1 Ответ

2 голосов
/ 29 января 2020

Боюсь, это ошибка. Я создал https://issues.apache.org/jira/browse/FLINK-15801 для его отслеживания.

Вы сможете обойти это, если измените одно из имен полей в ваших определениях времени строки. Измените либо имя логического поля: 

.field("timeAttribute", Types.SQL_TIMESTAMP)
    .rowtime(new Rowtime().timestampsFromField("rowtime").watermarksPeriodicBounded(2000))

, либо физическое поле происхождения: 

.field("rowtime", Types.SQL_TIMESTAMP)
    .rowtime(new Rowtime().timestampsFromField("timestamp").watermarksPeriodicBounded(2000))
...