[
https://issues.apache.org/jira/browse/FLINK-15230?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16995372#comment-16995372
]
kevin edited comment on FLINK-15230 at 12/13/19 5:36 AM:
---------------------------------------------------------
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
EnvironmentSettings bsSettings =
EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
StreamTableEnvironment tableEnv =
StreamTableEnvironment.create(env,bsSettings);
String JSON_SCHEMA =
"{" +
" type: 'object'," +
" properties:
{" + " id: \{ type: 'integer' }
," +
" name: \{ type: 'string' }," +
" timestamp: \{ type: 'string', format: 'date-time' }," +
" strings: { type: 'array', items:
{ type: 'string' }
}," +
" nested: { " +
" type: 'object'," +
" properties:
{ " + " booleanField: \{ type: 'boolean' }
," +
" decimalField: \{ type: 'number' }" +
" }" +
" }" +
" }" +
"}";
tableEnv.connect(new Kafka()
.version("universal")
.topic("w001")
.property("zookeeper.connect", "192.168.0.12:2181")
.property("bootstrap.servers",
"192.168.0.13:9092,192.168.0.14:9092,192.168.0.15:9092")
.property("group.id", "w01")
.startFromLatest()
)
.withFormat(new Json().deriveSchema())
//jsonSchema(JSON_SCHEMA).failOnMissingField(false)
.withSchema(new Schema().field("id", Types.BIG_DEC).field("name",
Types.STRING).field("timestamp", Types.SQL_TIMESTAMP)
.field("strings", Types.OBJECT_ARRAY(Types.STRING))
.field("nested",Types.ROW_NAMED(new
String[]\{"booleanField","decimalField"},new TypeInformation[]
{ Types.BOOLEAN, Types.BIG_DEC }
)))
.inAppendMode()
.registerTableSource("test");
Table query = tableEnv.sqlQuery("select name,nested.booleanField from test");
tableEnv.toAppendStream(query, Row.class).print();
tableEnv.execute("streaming");
was (Author: wzxwf):
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
EnvironmentSettings bsSettings =
EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env,bsSettings);
String JSON_SCHEMA =
"{" +
" type: 'object'," +
" properties: {" +
" id: \{ type: 'integer' }," +
" name: \{ type: 'string' }," +
" timestamp: \{ type: 'string', format: 'date-time' }," +
" strings: \{ type: 'array', items: { type: 'string' } }," +
" nested: { " +
" type: 'object'," +
" properties: { " +
" booleanField: \{ type: 'boolean' }," +
" decimalField: \{ type: 'number' }" +
" }" +
" }" +
" }" +
"}";
tableEnv.connect(new Kafka()
.version("universal")
.topic("w001")
.property("zookeeper.connect", "192.168.0.12:2181")
.property("bootstrap.servers",
"192.168.0.13:9092,192.168.0.14:9092,192.168.0.15:9092")
.property("group.id", "w01")
.startFromLatest()
)
.withFormat(new Json().deriveSchema())
//jsonSchema(JSON_SCHEMA).failOnMissingField(false)
.withSchema(new Schema().field("id", Types.BIG_DEC).field("name",
Types.STRING).field("timestamp", Types.SQL_TIMESTAMP)
.field("strings", Types.OBJECT_ARRAY(Types.STRING))
.field("nested",Types.ROW_NAMED(new
String[]\{"booleanField","decimalField"},new TypeInformation[]{
Types.BOOLEAN,
Types.BIG_DEC
})))
.inAppendMode()
.registerTableSource("test");
Table query = tableEnv.sqlQuery("select name,nested.booleanField from test");
tableEnv.toAppendStream(query, Row.class).print();
tableEnv.execute("streaming");
> flink1.9.1 table API JSON schema array type exception
> -----------------------------------------------------
>
> Key: FLINK-15230
> URL: https://issues.apache.org/jira/browse/FLINK-15230
> Project: Flink
> Issue Type: Bug
> Components: Formats (JSON, Avro, Parquet, ORC, SequenceFile), Table
> SQL / Planner
> Affects Versions: 1.9.0, 1.9.1
> Environment: flink1.9.1
> Reporter: kevin
> Priority: Major
>
> strings: { type: 'array', items:
> { type: 'string' }
> }
> .field("strings", Types.OBJECT_ARRAY(Types.STRING))
> Exception in thread "main" org.apache.flink.table.api.ValidationException:
> Type LEGACY(BasicArrayTypeInfo<String>) of table field 'strings' does not
> match with type BasicArrayTypeInfo<String> of the field 'strings' of the
> TableSource return type.Exception in thread "main"
> org.apache.flink.table.api.ValidationException: Type
> LEGACY(BasicArrayTypeInfo<String>) of table field 'strings' does not match
> with type BasicArrayTypeInfo<String> of the field 'strings' of the
> TableSource return type. at
> org.apache.flink.table.planner.sources.TableSourceUtil$$anonfun$4.apply(TableSourceUtil.scala:121)
> at
> org.apache.flink.table.planner.sources.TableSourceUtil$$anonfun$4.apply(TableSourceUtil.scala:92)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
> at
> scala.collection.IndexedSeqOptimized$class.foreach(IndexedSeqOptimized.scala:33)
> at scala.collection.mutable.ArrayOps$ofRef.foreach(ArrayOps.scala:186) at
> scala.collection.TraversableLike$class.map(TraversableLike.scala:234) at
> scala.collection.mutable.ArrayOps$ofRef.map(ArrayOps.scala:186) at
> org.apache.flink.table.planner.sources.TableSourceUtil$.computeIndexMapping(TableSourceUtil.scala:92)
> at
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecTableSourceScan.translateToPlanInternal(StreamExecTableSourceScan.scala:100)
> at
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecTableSourceScan.translateToPlanInternal(StreamExecTableSourceScan.scala:55)
> at
> org.apache.flink.table.planner.plan.nodes.exec.ExecNode$class.translateToPlan(ExecNode.scala:54)
> at
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecTableSourceScan.translateToPlan(StreamExecTableSourceScan.scala:55)
> at
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecCalc.translateToPlanInternal(StreamExecCalc.scala:86)
> at
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecCalc.translateToPlanInternal(StreamExecCalc.scala:46)
> at
> org.apache.flink.table.planner.plan.nodes.exec.ExecNode$class.translateToPlan(ExecNode.scala:54)
> at
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecCalc.translateToPlan(StreamExecCalc.scala:46)
> at
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecSink.translateToTransformation(StreamExecSink.scala:185)
> at
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecSink.translateToPlanInternal(StreamExecSink.scala:154)
> at
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecSink.translateToPlanInternal(StreamExecSink.scala:50)
> at
> org.apache.flink.table.planner.plan.nodes.exec.ExecNode$class.translateToPlan(ExecNode.scala:54)
> at
> org.apache.flink.table.planner.plan.nodes.physical.stream.StreamExecSink.translateToPlan(StreamExecSink.scala:50)
> at
> org.apache.flink.table.planner.delegation.StreamPlanner$$anonfun$translateToPlan$1.apply(StreamPlanner.scala:61)
> at
> org.apache.flink.table.planner.delegation.StreamPlanner$$anonfun$translateToPlan$1.apply(StreamPlanner.scala:60)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
> at
> scala.collection.TraversableLike$$anonfun$map$1.apply(TraversableLike.scala:234)
> at scala.collection.Iterator$class.foreach(Iterator.scala:891) at
> scala.collection.AbstractIterator.foreach(Iterator.scala:1334) at
> scala.collection.IterableLike$class.foreach(IterableLike.scala:72) at
> scala.collection.AbstractIterable.foreach(Iterable.scala:54) at
> scala.collection.TraversableLike$class.map(TraversableLike.scala:234) at
> scala.collection.AbstractTraversable.map(Traversable.scala:104) at
> org.apache.flink.table.planner.delegation.StreamPlanner.translateToPlan(StreamPlanner.scala:60)
> at
> org.apache.flink.table.planner.delegation.PlannerBase.translate(PlannerBase.scala:149)
> at
> org.apache.flink.table.api.java.internal.StreamTableEnvironmentImpl.toDataStream(StreamTableEnvironmentImpl.java:319)
> at
> org.apache.flink.table.api.java.internal.StreamTableEnvironmentImpl.toAppendStream(StreamTableEnvironmentImpl.java:227)
> at
> org.apache.flink.table.api.java.internal.StreamTableEnvironmentImpl.toAppendStream(StreamTableEnvironmentImpl.java:218)
> at com.wf.flink.sql.TableSqlJson.main(TableSqlJson.java:64) at
> sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
> at
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> at java.lang.reflect.Method.invoke(Method.java:498) at
> com.intellij.rt.execution.application.AppMain.main(AppMain.java:147)
> use jsonSchema(JSON_SCHEMA).failOnMissingField(false)
> not use .field("strings", Types.OBJECT_ARRAY(Types.STRING))
> throw Exception in thread "main"
> org.apache.flink.runtime.client.JobExecutionException: Job execution
> failed.Exception in thread "main"
> org.apache.flink.runtime.client.JobExecutionException: Job execution failed.
> at
> org.apache.flink.runtime.jobmaster.JobResult.toJobExecutionResult(JobResult.java:146)
> at
> org.apache.flink.runtime.minicluster.MiniCluster.executeJobBlocking(MiniCluster.java:626)
> at
> org.apache.flink.streaming.api.environment.LocalStreamEnvironment.execute(LocalStreamEnvironment.java:117)
> at
> org.apache.flink.table.planner.delegation.StreamExecutor.execute(StreamExecutor.java:46)
> at
> org.apache.flink.table.api.internal.TableEnvironmentImpl.execute(TableEnvironmentImpl.java:410)
> at com.wf.flink.sql.TableSqlJson.main(TableSqlJson.java:65) at
> sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at
> sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
> at
> sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
> at java.lang.reflect.Method.invoke(Method.java:498) at
> com.intellij.rt.execution.application.AppMain.main(AppMain.java:147)Caused
> by: java.lang.Exception:
> org.apache.flink.streaming.runtime.tasks.ExceptionInChainedOperatorException:
> Could not forward element to next operator at
> org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.checkThrowSourceExecutionException(SourceStreamTask.java:217)
> at
> org.apache.flink.streaming.runtime.tasks.SourceStreamTask.processInput(SourceStreamTask.java:133)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.run(StreamTask.java:301)
> at
> org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:406)
> at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:705) at
> org.apache.flink.runtime.taskmanager.Task.run(Task.java:530) at
> java.lang.Thread.run(Thread.java:745)Caused by:
> org.apache.flink.streaming.runtime.tasks.ExceptionInChainedOperatorException:
> Could not forward element to next operator at
> org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:651)
> at
> org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:612)
> at
> org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.collect(OperatorChain.java:592)
> at
> org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:727)
> at
> org.apache.flink.streaming.api.operators.AbstractStreamOperator$CountingOutput.collect(AbstractStreamOperator.java:705)
> at
> org.apache.flink.streaming.api.operators.StreamSourceContexts$NonTimestampContext.collect(StreamSourceContexts.java:104)
> at
> org.apache.flink.streaming.api.operators.StreamSourceContexts$NonTimestampContext.collectWithTimestamp(StreamSourceContexts.java:111)
> at
> org.apache.flink.streaming.connectors.kafka.internals.AbstractFetcher.emitRecordWithTimestamp(AbstractFetcher.java:398)
> at
> org.apache.flink.streaming.connectors.kafka.internal.KafkaFetcher.emitRecord(KafkaFetcher.java:185)
> at
> org.apache.flink.streaming.connectors.kafka.internal.KafkaFetcher.runFetchLoop(KafkaFetcher.java:150)
> at
> org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.run(FlinkKafkaConsumerBase.java:715)
> at
> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:100)
> at
> org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:63)
> at
> org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:203)Caused
> by: java.lang.ClassCastException: [Ljava.lang.Object; cannot be cast to
> [Ljava.lang.String; at
> org.apache.flink.api.common.typeutils.base.array.StringArraySerializer.copy(StringArraySerializer.java:35)
> at
> org.apache.flink.api.java.typeutils.runtime.RowSerializer.copy(RowSerializer.java:93)
> at
> org.apache.flink.api.java.typeutils.runtime.RowSerializer.copy(RowSerializer.java:44)
> at
> org.apache.flink.streaming.runtime.tasks.OperatorChain$CopyingChainingOutput.pushToOperator(OperatorChain.java:635)
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
