Hello Thanks to any Flink expert here who can help me take m Apache Flink #troubleshooting

Hello! Thanks to any Flink expert here who can hel...

Konstantinos Samaras-Tsakiris

03/10/2023, 1:57 PM

Hello! Thanks to any Flink expert here who can help me take my first baby steps with this system!! 🙇 🙏 I have a basic question about ROWTIME. I'm using the pyflink Table API and... 1. define a streaming table that has a column

deviceTimestamp

as ROWTIME:

Copy code

streamTableEnv.create_table("source",
  .schema(Schema.new_builder()
    .column('deviceTimestamp' ...)
    ...
    .watermark('deviceTimestamp', "deviceTimestamp - INTERVAL '0.04' SECOND")

2. then join it with the output of a UDTF that adds a new

sampleTimestamp

column 3. then try to perform a window operation on the

sampleTimestamp

column I realize that I can't do this because windowing can only happen on the ROWTIME column. Do you know how I can transform the ROWTIME column, to be

sampleTimestamp

instead of

deviceTimestamp

Konstantinos Samaras-Tsakiris

03/10/2023, 2:01 PM

I also tried to put the data into an intermediate stream:

Copy code

streamTableEnv.create_temporary_table("splatSamples",
  .schema(Schema.new_builder()
    .column('deviceTimestamp' ...)
    .column('sampleTimestamp' ...)
    ...
    .watermark('sampleTimestamp', "sampleTimestamp - INTERVAL '0.04' SECOND")

streamTableEnv.from_path("source")
  .join_lateral(my_udtf, ...)
  .execute_insert("splatSamples")

streamTableEnv.from_path("splatSamples")
  .window(Slide ...
    .on(col("sampleTimestamp"))

The problem with this approach is that I don't know what connector to use for the

splatSamples

table. I don't need the data to exit Flink. Is there a way to define this intermediate table without a connector?

Dian Fu

03/11/2023, 1:48 PM

@Konstantinos Samaras-Tsakiris Hey, I guess you could try the following approach: • Convert the Table of the joined result of step 2 to DataStream • Convert the DataStream back to Table and declare

sampleTimestamp

as ROWTIME column [1] and perform window operation on it [1] https://github.com/apache/flink/blob/master/flink-python/pyflink/examples/table/windowing/tumble_window.py#L49

Konstantinos Samaras-Tsakiris

03/11/2023, 2:58 PM

Amazing! @Dian Fu thanks a lot for the hint, I'll try this. I'll post the result.

5 Views

Open in Slack

Previous Next