Re: Avro UTF-8 deserialization on integration DataStreams API -> Table API (with Confluent SchemaRegistry)
Update:
I assume you are talking about DataStreamSource.process(.), right?
(
https://nightlies.apache.org/flink/flink-docs-release-1.14/api/java/org/apache/flink/streaming/api/datastream/DataStream.html#process-org.apache.flink.streaming.api.functions.ProcessFunction-
)
So similar to a .map(.) function you suggest to touch each and every event
and cast the `CharSequence` to a `String`?
I also thought about that workaround but for 2 reasons I don't really like
it:
1) I do not really want to introduce another pojo which only lives inside
Flink runtime, also my record is quite huge and nested -> additional
maintenance for 2 (actually equal) data structures
2) I do not want to add one more step to the pipeline and transform each
and every record -> produces overhead
Thats why I was looking for a solution that transfers the data stream
records in the avro pojo - with UTF8s - to the table API and represents the
concerning as `STRING` (as UTF8 is an analogous data type to STRING, both
based on `CharSequence`)
Thanks & Best
Peter
On Tue, Oct 19, 2021 at 4:47 AM Caizhi Weng wrote:
> Hi!
>
> You can call streamSource.processRecord to change the CharSequence to a
> String, then change the stream to a table.
>
> Peter Schrott 于2021年10月18日周一 下午8:40写道：
>
>> Hi there,
>>
>> I have a Kafka topic where the schema of its values is defined by the
>> "MyRecord" record in the following Avro IDL and registered to the
>> Confluent
>> Schema Registry:
>>
>> @namespace("org.example")
>> protocol MyProtocol {
>>record MyRecord {
>> string text;
>>}
>> }
>>
>> The topic is consumed with a KafkaSource and then then passed into
>> StreamTableEnvironment. On the temporary view I want to run SQL queries.
>>
>> But the following exception is thrown on startup of the job:
>>
>> Exception in thread "main" org.apache.flink.table.api.ValidationException:
>> SQL validation failed. From line 0, column 0 to line 1, column 58: Cannot
>> apply 'LIKE' to arguments of type 'LIKE(> '...')>, )'. Supported form(s): 'LIKE(, , )'
>> at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org
>>
Re: Avro UTF-8 deserialization on integration DataStreams API -> Table API (with Confluent SchemaRegistry)
Hi!
Sorry for misleading. I mean DataStream#process, see
https://nightlies.apache.org/flink/flink-docs-release-1.14/api/java/org/apache/flink/streaming/api/datastream/DataStream.html#process-org.apache.flink.streaming.api.functions.ProcessFunction-
Peter Schrott 于2021年10月19日周二 下午3:10写道：
> Hi & thanks!
>
> DataStreamSource does not provide a method processRecord:
>
> https://nightlies.apache.org/flink/flink-docs-release-1.14/api/java/org/apache/flink/streaming/api/datastream/DataStreamSource.html
>
> Can you point me to the docs for that?
>
> Thanks, Peter
>
> On Tue, Oct 19, 2021 at 4:47 AM Caizhi Weng wrote:
>
>> Hi!
>>
>> You can call streamSource.processRecord to change the CharSequence to a
>> String, then change the stream to a table.
>>
>> Peter Schrott 于2021年10月18日周一 下午8:40写道：
>>
>>> Hi there,
>>>
>>> I have a Kafka topic where the schema of its values is defined by the
>>> "MyRecord" record in the following Avro IDL and registered to the
>>> Confluent
>>> Schema Registry:
>>>
>>> @namespace("org.example")
>>> protocol MyProtocol {
>>>record MyRecord {
>>> string text;
>>>}
>>> }
>>>
>>> The topic is consumed with a KafkaSource and then then passed into
>>> StreamTableEnvironment. On the temporary view I want to run SQL queries.
>>>
>>> But the following exception is thrown on startup of the job:
>>>
>>> Exception in thread "main" org.apache.flink.table.api.ValidationException:
>>> SQL validation failed. From line 0, column 0 to line 1, column 58: Cannot
>>> apply 'LIKE' to arguments of type 'LIKE(>> '...')>, )'. Supported form(s): 'LIKE(, ,
>>> )'
>>> at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org
>>>
Re: Avro UTF-8 deserialization on integration DataStreams API -> Table API (with Confluent SchemaRegistry)
Hi & thanks!
DataStreamSource does not provide a method processRecord:
https://nightlies.apache.org/flink/flink-docs-release-1.14/api/java/org/apache/flink/streaming/api/datastream/DataStreamSource.html
Can you point me to the docs for that?
Thanks, Peter
On Tue, Oct 19, 2021 at 4:47 AM Caizhi Weng wrote:
> Hi!
>
> You can call streamSource.processRecord to change the CharSequence to a
> String, then change the stream to a table.
>
> Peter Schrott 于2021年10月18日周一 下午8:40写道：
>
>> Hi there,
>>
>> I have a Kafka topic where the schema of its values is defined by the
>> "MyRecord" record in the following Avro IDL and registered to the
>> Confluent
>> Schema Registry:
>>
>> @namespace("org.example")
>> protocol MyProtocol {
>>record MyRecord {
>> string text;
>>}
>> }
>>
>> The topic is consumed with a KafkaSource and then then passed into
>> StreamTableEnvironment. On the temporary view I want to run SQL queries.
>>
>> But the following exception is thrown on startup of the job:
>>
>> Exception in thread "main" org.apache.flink.table.api.ValidationException:
>> SQL validation failed. From line 0, column 0 to line 1, column 58: Cannot
>> apply 'LIKE' to arguments of type 'LIKE(> '...')>, )'. Supported form(s): 'LIKE(, , )'
>> at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org
>>
Re: Avro UTF-8 deserialization on integration DataStreams API -> Table API (with Confluent SchemaRegistry)
Hi!
You can call streamSource.processRecord to change the CharSequence to a
String, then change the stream to a table.
Peter Schrott 于2021年10月18日周一 下午8:40写道：
> Hi there,
>
> I have a Kafka topic where the schema of its values is defined by the
> "MyRecord" record in the following Avro IDL and registered to the Confluent
> Schema Registry:
>
> @namespace("org.example")
> protocol MyProtocol {
>record MyRecord {
> string text;
>}
> }
>
> The topic is consumed with a KafkaSource and then then passed into
> StreamTableEnvironment. On the temporary view I want to run SQL queries.
>
> But the following exception is thrown on startup of the job:
>
> Exception in thread "main" org.apache.flink.table.api.ValidationException:
> SQL validation failed. From line 0, column 0 to line 1, column 58: Cannot
> apply 'LIKE' to arguments of type 'LIKE( '...')>, )'. Supported form(s): 'LIKE(, , )'
> at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org
>
Avro UTF-8 deserialization on integration DataStreams API -> Table API (with Confluent SchemaRegistry)
Hi there,
I have a Kafka topic where the schema of its values is defined by the
"MyRecord" record in the following Avro IDL and registered to the Confluent
Schema Registry:
@namespace("org.example")
protocol MyProtocol {
record MyRecord {
string text;
}
}
The topic is consumed with a KafkaSource and then then passed into
StreamTableEnvironment. On the temporary view I want to run SQL queries.
But the following exception is thrown on startup of the job:
Exception in thread "main"
org.apache.flink.table.api.ValidationException: SQL validation failed.
>From line 0, column 0 to line 1, column 58: Cannot apply 'LIKE' to
arguments of type 'LIKE(,
)'. Supported form(s): 'LIKE(, , )'
at org.apache.flink.table.planner.calcite.FlinkPlannerImpl.org
