Re: Define rowtime on intermediate table field

[Yun Gao]

Hi Sumeet,

I think you might first convert the table back to the DataStream [1],
then define the timestamp and watermark with 
`assignTimestampsAndWatermarks(...)`, 
and then convert it back to table[2].

Best,
Yun

[1] 
https://ci.apache.org/projects/flink/flink-docs-master/docs/dev/table/common/#convert-a-table-into-a-datastream
[2] 
https://ci.apache.org/projects/flink/flink-docs-master/docs/dev/table/concepts/time_attributes/#during-datastream-to-table-conversion



 --Original Mail --
Sender:Sumeet Malhotra 
Send Date:Tue May 4 16:32:10 2021
Recipients:user 
Subject:Define rowtime on intermediate table field

Hi,

My use case involves reading raw data records from Kafka and processing them. 
The records are coming from a database, where a periodic job reads new rows, 
packages them into a single JSON object (as described below) and writes the 
entire record to Kafka.

{
'id': 'some_id',
'key_a': 'value_a',
'key_b': 'value_b',
'result': {
'columns': [
'col_a',
'col_b',
'col_c',
'col_d'
],
'rows': [
['2021-05-04T05:23:13.953610Z', '655361', '8013', '0'],
['2021-05-04T05:23:13.953610Z', '655362', '4000', '456'],
['2021-05-04T05:23:13.953610Z', '655363', '2', '562'],
...
...
]
}
}

As can be seen, the row time is actually embedded in the `result` object.

What I'm doing at the moment is to run this data through a user defined table 
function, which parses the `result` object as a string, and emits multiple rows 
that include the timestamp field. This is working fine.

In the next step, I would want to perform windowing on this transformed data. 
That requires defining the event time attribute along with the watermark. As I 
understand, this can be done either during the initial table DDL definition or 
during conversion to a datastream.

Since I extract the timestamp value only after reading from Kafka, how can I 
define an event time attribute on the intermediate table that's basically a 
result of the user defined table function?

The only solution I can think of at the moment, is to write the intermediate 
table back to Kafka, and then create a new consumer that reads from Kafka, 
where I can define the event time attribute as part of its DDL. This most 
likely won't be good performance wise. I'm looking at any other way, I can 
define event time on results of my user defined table function?

Thanks in advance,
Sumeet

Define rowtime on intermediate table field

[Sumeet Malhotra]

Hi,

My use case involves reading raw data records from Kafka and processing
them. The records are coming from a database, where a periodic job reads
new rows, packages them into a single JSON object (as described below) and
writes the entire record to Kafka.

{
'id': 'some_id',
'key_a': 'value_a',
'key_b': 'value_b',
'result': {
'columns': [
'col_a',
'col_b',
'col_c',
'col_d'
],
'rows': [
['2021-05-04T05:23:13.953610Z', '655361', '8013', '0'],
['2021-05-04T05:23:13.953610Z', '655362', '4000', '456'],
['2021-05-04T05:23:13.953610Z', '655363', '2', '562'],
...
...
]
}
}

As can be seen, the row time is actually embedded in the `result` object.

What I'm doing at the moment is to run this data through a user defined
table function, which parses the `result` object as a string, and emits
multiple rows that include the timestamp field. This is working fine.

In the next step, I would want to perform windowing on this transformed
data. That requires defining the event time attribute along with the
watermark. As I understand, this can be done either during the initial
table DDL definition or during conversion to a datastream.

Since I extract the timestamp value only after reading from Kafka, how can
I define an event time attribute on the intermediate table that's basically
a result of the user defined table function?

The only solution I can think of at the moment, is to write the
intermediate table back to Kafka, and then create a new consumer that reads
from Kafka, where I can define the event time attribute as part of its DDL.
This most likely won't be good performance wise. I'm looking at any other
way, I can define event time on results of my user defined table function?

Thanks in advance,
Sumeet