You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by Guoqin Zheng <la...@gmail.com> on 2021/11/15 17:42:53 UTC
PyFlink SQL window aggregation data not written out to file
Hi all,
I am reposting my stackoverflow question
<https://stackoverflow.com/questions/69971277/flink-sql-tumble-aggregation-result-not-written-out-to-filesystem-locally>
here.
So I would like to do a local test of my pyflink sql application. The sql
query does a very simple job: read from source, window aggregation and
write result to a sink.
In the production, the source and sink will be kinesis stream. But since I
need to do a local test, I am mocking out the source and sink with the
local filesystem.
The problem is the local job always runs successfully, but the results were
never written out to the sink file. Curious what I am missing here.
Note:
- I am using Flink 1.11.0
- if I directly read data from the source table by select query and
write to sink without windowing or grouping, it works just fine. That means
the source and sink table is set up correctly. So it seems the problem is
around the Tumbling and grouping for the local filesystem.
- This code works fine with Kinesis source and sink.
Here is the SQL code:
```
CREATE TABLE incoming_data (
requestId VARCHAR(4),
groupId VARCHAR(32),
userId VARCHAR(32),
requestStartTime VARCHAR(32),
processTime AS PROCTIME(),
requestTime AS TO_TIMESTAMP(SUBSTR(REPLACE(requestStartTime,
'T', ' '), 0, 23), 'yyyy-MM-dd HH:mm:ss.SSS'),
WATERMARK FOR requestTime AS requestTime - INTERVAL '5' SECOND
) WITH (
'connector' = 'filesystem',
'path' = '/path/to/test/json/file.json',
'format' = 'json',
'json.timestamp-format.standard' = 'ISO-8601'
)
CREATE TABLE user_latest_request (
groupId VARCHAR(32),
userId VARCHAR(32),
latestRequestTime TIMESTAMP
) WITH (
'connector' = 'filesystem',
'path' = '/path/to/sink',
'format' = 'csv')
INSERT INTO user_latest_request
SELECT groupId,
userId,
MAX(requestTime) as latestRequestTime
FROM incoming_data
GROUP BY TUMBLE(processTime, INTERVAL '1' SECOND), groupId, userId;
```
Thanks!
Re: PyFlink SQL window aggregation data not written out to file
Posted by Guoqin Zheng <la...@gmail.com>.
Hi Roman,
Thanks for the update and testing locally. This is very informative.
-Guoqin
On Mon, Nov 22, 2021 at 10:55 AM Roman Khachatryan <ro...@apache.org> wrote:
> Hi Guoqin,
>
> I was able to reproduce the problem locally. I can see that at the
> time of window firing the services are already closed because of the
> emitted MAX_WATERMARK.
>
> Previously, there were some discussions around waiting for all timers
> to complete [1], but AFAIK there was not much demand to implement it.
> Probably Piotr or Arvid know more.
>
> Please feel free to bump the ticket.
>
> [1]
> https://issues.apache.org/jira/browse/FLINK-18647
>
> Regards,
> Roman
>
> On Wed, Nov 17, 2021 at 7:36 PM Guoqin Zheng <la...@gmail.com>
> wrote:
> >
> > Hi Roman,
> >
> > Thanks for the detailed explanation.
> >
> > I did try 1.13 and 1.14, but it still didn't work.
> >
> > I explicitly enabled the checkpoint with:
> `env.enable_checkpointing(10)`. Any other configurations I need to set?
> >
> > Thanks,
> > -Guoqin
> >
> > On Wed, Nov 17, 2021 at 4:30 AM Roman Khachatryan <ro...@apache.org>
> wrote:
> >>
> >> Hi Guoqin,
> >>
> >> Thanks for the clarification.
> >>
> >> Processing time windows actually don't need watermarks: they fire when
> >> window end time comes.
> >> But the job will likely finish earlier because of the bounded input.
> >>
> >> Handling of this case was improved in 1.14 as part of FLIP-147, as
> >> well as in previous versions.
> >> So I'd suggest trying the most recent Flink version.
> >>
> >> Regards,
> >> Roman
> >>
> >> On Mon, Nov 15, 2021 at 10:17 PM Guoqin Zheng <la...@gmail.com>
> wrote:
> >> >
> >> > Hi Roman,
> >> >
> >> > Thanks for your quick response!
> >> >
> >> > Yes, it does seem to be the window closing problem.
> >> >
> >> > So if I change the tumble window on eventTime, which is column
> 'requestTime', it works fine. I guess the EOF of the test data file kicks
> in a watermark of Long.MAX_VALUE.
> >> >
> >> > But my application code needs to calculate the result based on the
> tumble window of processTime. Curious what is the typical setup for such a
> local test. ie. How do we inform the Flink to close the window if the input
> stream hits the end.
> >> >
> >> > Thanks,
> >> > -Guoqin
> >> >
> >> > On Mon, Nov 15, 2021 at 11:21 AM Roman Khachatryan <ro...@apache.org>
> wrote:
> >> >>
> >> >> Hi Guoqin,
> >> >>
> >> >> I think the problem might be related to watermarks and checkpointing:
> >> >> - if the file is too small, the only watermark will be the one after
> >> >> fully reading the file
> >> >> - in exactly once mode, sink waits for a checkpoint completion before
> >> >> committing the files
> >> >>
> >> >> Recently, there were some improvements in handling the last
> >> >> checkpoint, so you might want to try Flink 1.14.
> >> >> You might also try to configure checkpointing (e.g. decrease interval
> >> >> [1]) and watermark generation [2] to diagnose the problem.
> >> >>
> >> >> [1]
> https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/deployment/config/#execution-checkpointing-interval
> >> >> [2]
> https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/sql/create/#watermark
> >> >>
> >> >>
> >> >> Regards,
> >> >> Roman
> >> >>
> >> >> On Mon, Nov 15, 2021 at 6:43 PM Guoqin Zheng <la...@gmail.com>
> wrote:
> >> >> >
> >> >> > Hi all,
> >> >> >
> >> >> > I am reposting my stackoverflow question here.
> >> >> >
> >> >> > So I would like to do a local test of my pyflink sql application.
> The sql query does a very simple job: read from source, window aggregation
> and write result to a sink.
> >> >> >
> >> >> > In the production, the source and sink will be kinesis stream. But
> since I need to do a local test, I am mocking out the source and sink with
> the local filesystem.
> >> >> >
> >> >> > The problem is the local job always runs successfully, but the
> results were never written out to the sink file. Curious what I am missing
> here.
> >> >> >
> >> >> > Note:
> >> >> >
> >> >> > I am using Flink 1.11.0
> >> >> > if I directly read data from the source table by select query and
> write to sink without windowing or grouping, it works just fine. That means
> the source and sink table is set up correctly. So it seems the problem is
> around the Tumbling and grouping for the local filesystem.
> >> >> > This code works fine with Kinesis source and sink.
> >> >> >
> >> >> >
> >> >> > Here is the SQL code:
> >> >> >
> >> >> > ```
> >> >> >
> >> >> > CREATE TABLE incoming_data (
> >> >> > requestId VARCHAR(4),
> >> >> > groupId VARCHAR(32),
> >> >> > userId VARCHAR(32),
> >> >> > requestStartTime VARCHAR(32),
> >> >> > processTime AS PROCTIME(),
> >> >> > requestTime AS
> TO_TIMESTAMP(SUBSTR(REPLACE(requestStartTime, 'T', ' '), 0, 23),
> 'yyyy-MM-dd HH:mm:ss.SSS'),
> >> >> > WATERMARK FOR requestTime AS requestTime - INTERVAL '5'
> SECOND
> >> >> > ) WITH (
> >> >> > 'connector' = 'filesystem',
> >> >> > 'path' = '/path/to/test/json/file.json',
> >> >> > 'format' = 'json',
> >> >> > 'json.timestamp-format.standard' = 'ISO-8601'
> >> >> > )
> >> >> >
> >> >> > CREATE TABLE user_latest_request (
> >> >> > groupId VARCHAR(32),
> >> >> > userId VARCHAR(32),
> >> >> > latestRequestTime TIMESTAMP
> >> >> > ) WITH (
> >> >> > 'connector' = 'filesystem',
> >> >> > 'path' = '/path/to/sink',
> >> >> > 'format' = 'csv'
> >> >> > )
> >> >> >
> >> >> > INSERT INTO user_latest_request
> >> >> > SELECT groupId,
> >> >> > userId,
> >> >> > MAX(requestTime) as latestRequestTime
> >> >> > FROM incoming_data
> >> >> >
> >> >> > GROUP BY TUMBLE(processTime, INTERVAL '1' SECOND), groupId,
> userId;
> >> >> >
> >> >> > ```
> >> >> >
> >> >> >
> >> >> > Thanks!
>
Re: PyFlink SQL window aggregation data not written out to file
Posted by Roman Khachatryan <ro...@apache.org>.
Hi Guoqin,
I was able to reproduce the problem locally. I can see that at the
time of window firing the services are already closed because of the
emitted MAX_WATERMARK.
Previously, there were some discussions around waiting for all timers
to complete [1], but AFAIK there was not much demand to implement it.
Probably Piotr or Arvid know more.
Please feel free to bump the ticket.
[1]
https://issues.apache.org/jira/browse/FLINK-18647
Regards,
Roman
On Wed, Nov 17, 2021 at 7:36 PM Guoqin Zheng <la...@gmail.com> wrote:
>
> Hi Roman,
>
> Thanks for the detailed explanation.
>
> I did try 1.13 and 1.14, but it still didn't work.
>
> I explicitly enabled the checkpoint with: `env.enable_checkpointing(10)`. Any other configurations I need to set?
>
> Thanks,
> -Guoqin
>
> On Wed, Nov 17, 2021 at 4:30 AM Roman Khachatryan <ro...@apache.org> wrote:
>>
>> Hi Guoqin,
>>
>> Thanks for the clarification.
>>
>> Processing time windows actually don't need watermarks: they fire when
>> window end time comes.
>> But the job will likely finish earlier because of the bounded input.
>>
>> Handling of this case was improved in 1.14 as part of FLIP-147, as
>> well as in previous versions.
>> So I'd suggest trying the most recent Flink version.
>>
>> Regards,
>> Roman
>>
>> On Mon, Nov 15, 2021 at 10:17 PM Guoqin Zheng <la...@gmail.com> wrote:
>> >
>> > Hi Roman,
>> >
>> > Thanks for your quick response!
>> >
>> > Yes, it does seem to be the window closing problem.
>> >
>> > So if I change the tumble window on eventTime, which is column 'requestTime', it works fine. I guess the EOF of the test data file kicks in a watermark of Long.MAX_VALUE.
>> >
>> > But my application code needs to calculate the result based on the tumble window of processTime. Curious what is the typical setup for such a local test. ie. How do we inform the Flink to close the window if the input stream hits the end.
>> >
>> > Thanks,
>> > -Guoqin
>> >
>> > On Mon, Nov 15, 2021 at 11:21 AM Roman Khachatryan <ro...@apache.org> wrote:
>> >>
>> >> Hi Guoqin,
>> >>
>> >> I think the problem might be related to watermarks and checkpointing:
>> >> - if the file is too small, the only watermark will be the one after
>> >> fully reading the file
>> >> - in exactly once mode, sink waits for a checkpoint completion before
>> >> committing the files
>> >>
>> >> Recently, there were some improvements in handling the last
>> >> checkpoint, so you might want to try Flink 1.14.
>> >> You might also try to configure checkpointing (e.g. decrease interval
>> >> [1]) and watermark generation [2] to diagnose the problem.
>> >>
>> >> [1] https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/deployment/config/#execution-checkpointing-interval
>> >> [2] https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/sql/create/#watermark
>> >>
>> >>
>> >> Regards,
>> >> Roman
>> >>
>> >> On Mon, Nov 15, 2021 at 6:43 PM Guoqin Zheng <la...@gmail.com> wrote:
>> >> >
>> >> > Hi all,
>> >> >
>> >> > I am reposting my stackoverflow question here.
>> >> >
>> >> > So I would like to do a local test of my pyflink sql application. The sql query does a very simple job: read from source, window aggregation and write result to a sink.
>> >> >
>> >> > In the production, the source and sink will be kinesis stream. But since I need to do a local test, I am mocking out the source and sink with the local filesystem.
>> >> >
>> >> > The problem is the local job always runs successfully, but the results were never written out to the sink file. Curious what I am missing here.
>> >> >
>> >> > Note:
>> >> >
>> >> > I am using Flink 1.11.0
>> >> > if I directly read data from the source table by select query and write to sink without windowing or grouping, it works just fine. That means the source and sink table is set up correctly. So it seems the problem is around the Tumbling and grouping for the local filesystem.
>> >> > This code works fine with Kinesis source and sink.
>> >> >
>> >> >
>> >> > Here is the SQL code:
>> >> >
>> >> > ```
>> >> >
>> >> > CREATE TABLE incoming_data (
>> >> > requestId VARCHAR(4),
>> >> > groupId VARCHAR(32),
>> >> > userId VARCHAR(32),
>> >> > requestStartTime VARCHAR(32),
>> >> > processTime AS PROCTIME(),
>> >> > requestTime AS TO_TIMESTAMP(SUBSTR(REPLACE(requestStartTime, 'T', ' '), 0, 23), 'yyyy-MM-dd HH:mm:ss.SSS'),
>> >> > WATERMARK FOR requestTime AS requestTime - INTERVAL '5' SECOND
>> >> > ) WITH (
>> >> > 'connector' = 'filesystem',
>> >> > 'path' = '/path/to/test/json/file.json',
>> >> > 'format' = 'json',
>> >> > 'json.timestamp-format.standard' = 'ISO-8601'
>> >> > )
>> >> >
>> >> > CREATE TABLE user_latest_request (
>> >> > groupId VARCHAR(32),
>> >> > userId VARCHAR(32),
>> >> > latestRequestTime TIMESTAMP
>> >> > ) WITH (
>> >> > 'connector' = 'filesystem',
>> >> > 'path' = '/path/to/sink',
>> >> > 'format' = 'csv'
>> >> > )
>> >> >
>> >> > INSERT INTO user_latest_request
>> >> > SELECT groupId,
>> >> > userId,
>> >> > MAX(requestTime) as latestRequestTime
>> >> > FROM incoming_data
>> >> >
>> >> > GROUP BY TUMBLE(processTime, INTERVAL '1' SECOND), groupId, userId;
>> >> >
>> >> > ```
>> >> >
>> >> >
>> >> > Thanks!
Re: PyFlink SQL window aggregation data not written out to file
Posted by Guoqin Zheng <la...@gmail.com>.
Hi Roman,
Thanks for the detailed explanation.
I did try 1.13 and 1.14, but it still didn't work.
I explicitly enabled the checkpoint with:
`env.enable_checkpointing(10)`. Any other configurations I need to set?
Thanks,
-Guoqin
On Wed, Nov 17, 2021 at 4:30 AM Roman Khachatryan <ro...@apache.org> wrote:
> Hi Guoqin,
>
> Thanks for the clarification.
>
> Processing time windows actually don't need watermarks: they fire when
> window end time comes.
> But the job will likely finish earlier because of the bounded input.
>
> Handling of this case was improved in 1.14 as part of FLIP-147, as
> well as in previous versions.
> So I'd suggest trying the most recent Flink version.
>
> Regards,
> Roman
>
> On Mon, Nov 15, 2021 at 10:17 PM Guoqin Zheng <la...@gmail.com>
> wrote:
> >
> > Hi Roman,
> >
> > Thanks for your quick response!
> >
> > Yes, it does seem to be the window closing problem.
> >
> > So if I change the tumble window on eventTime, which is column
> 'requestTime', it works fine. I guess the EOF of the test data file kicks
> in a watermark of Long.MAX_VALUE.
> >
> > But my application code needs to calculate the result based on the
> tumble window of processTime. Curious what is the typical setup for such a
> local test. ie. How do we inform the Flink to close the window if the input
> stream hits the end.
> >
> > Thanks,
> > -Guoqin
> >
> > On Mon, Nov 15, 2021 at 11:21 AM Roman Khachatryan <ro...@apache.org>
> wrote:
> >>
> >> Hi Guoqin,
> >>
> >> I think the problem might be related to watermarks and checkpointing:
> >> - if the file is too small, the only watermark will be the one after
> >> fully reading the file
> >> - in exactly once mode, sink waits for a checkpoint completion before
> >> committing the files
> >>
> >> Recently, there were some improvements in handling the last
> >> checkpoint, so you might want to try Flink 1.14.
> >> You might also try to configure checkpointing (e.g. decrease interval
> >> [1]) and watermark generation [2] to diagnose the problem.
> >>
> >> [1]
> https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/deployment/config/#execution-checkpointing-interval
> >> [2]
> https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/sql/create/#watermark
> >>
> >>
> >> Regards,
> >> Roman
> >>
> >> On Mon, Nov 15, 2021 at 6:43 PM Guoqin Zheng <la...@gmail.com>
> wrote:
> >> >
> >> > Hi all,
> >> >
> >> > I am reposting my stackoverflow question here.
> >> >
> >> > So I would like to do a local test of my pyflink sql application. The
> sql query does a very simple job: read from source, window aggregation and
> write result to a sink.
> >> >
> >> > In the production, the source and sink will be kinesis stream. But
> since I need to do a local test, I am mocking out the source and sink with
> the local filesystem.
> >> >
> >> > The problem is the local job always runs successfully, but the
> results were never written out to the sink file. Curious what I am missing
> here.
> >> >
> >> > Note:
> >> >
> >> > I am using Flink 1.11.0
> >> > if I directly read data from the source table by select query and
> write to sink without windowing or grouping, it works just fine. That means
> the source and sink table is set up correctly. So it seems the problem is
> around the Tumbling and grouping for the local filesystem.
> >> > This code works fine with Kinesis source and sink.
> >> >
> >> >
> >> > Here is the SQL code:
> >> >
> >> > ```
> >> >
> >> > CREATE TABLE incoming_data (
> >> > requestId VARCHAR(4),
> >> > groupId VARCHAR(32),
> >> > userId VARCHAR(32),
> >> > requestStartTime VARCHAR(32),
> >> > processTime AS PROCTIME(),
> >> > requestTime AS TO_TIMESTAMP(SUBSTR(REPLACE(requestStartTime,
> 'T', ' '), 0, 23), 'yyyy-MM-dd HH:mm:ss.SSS'),
> >> > WATERMARK FOR requestTime AS requestTime - INTERVAL '5' SECOND
> >> > ) WITH (
> >> > 'connector' = 'filesystem',
> >> > 'path' = '/path/to/test/json/file.json',
> >> > 'format' = 'json',
> >> > 'json.timestamp-format.standard' = 'ISO-8601'
> >> > )
> >> >
> >> > CREATE TABLE user_latest_request (
> >> > groupId VARCHAR(32),
> >> > userId VARCHAR(32),
> >> > latestRequestTime TIMESTAMP
> >> > ) WITH (
> >> > 'connector' = 'filesystem',
> >> > 'path' = '/path/to/sink',
> >> > 'format' = 'csv'
> >> > )
> >> >
> >> > INSERT INTO user_latest_request
> >> > SELECT groupId,
> >> > userId,
> >> > MAX(requestTime) as latestRequestTime
> >> > FROM incoming_data
> >> >
> >> > GROUP BY TUMBLE(processTime, INTERVAL '1' SECOND), groupId,
> userId;
> >> >
> >> > ```
> >> >
> >> >
> >> > Thanks!
>
Re: PyFlink SQL window aggregation data not written out to file
Posted by Roman Khachatryan <ro...@apache.org>.
Hi Guoqin,
Thanks for the clarification.
Processing time windows actually don't need watermarks: they fire when
window end time comes.
But the job will likely finish earlier because of the bounded input.
Handling of this case was improved in 1.14 as part of FLIP-147, as
well as in previous versions.
So I'd suggest trying the most recent Flink version.
Regards,
Roman
On Mon, Nov 15, 2021 at 10:17 PM Guoqin Zheng <la...@gmail.com> wrote:
>
> Hi Roman,
>
> Thanks for your quick response!
>
> Yes, it does seem to be the window closing problem.
>
> So if I change the tumble window on eventTime, which is column 'requestTime', it works fine. I guess the EOF of the test data file kicks in a watermark of Long.MAX_VALUE.
>
> But my application code needs to calculate the result based on the tumble window of processTime. Curious what is the typical setup for such a local test. ie. How do we inform the Flink to close the window if the input stream hits the end.
>
> Thanks,
> -Guoqin
>
> On Mon, Nov 15, 2021 at 11:21 AM Roman Khachatryan <ro...@apache.org> wrote:
>>
>> Hi Guoqin,
>>
>> I think the problem might be related to watermarks and checkpointing:
>> - if the file is too small, the only watermark will be the one after
>> fully reading the file
>> - in exactly once mode, sink waits for a checkpoint completion before
>> committing the files
>>
>> Recently, there were some improvements in handling the last
>> checkpoint, so you might want to try Flink 1.14.
>> You might also try to configure checkpointing (e.g. decrease interval
>> [1]) and watermark generation [2] to diagnose the problem.
>>
>> [1] https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/deployment/config/#execution-checkpointing-interval
>> [2] https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/sql/create/#watermark
>>
>>
>> Regards,
>> Roman
>>
>> On Mon, Nov 15, 2021 at 6:43 PM Guoqin Zheng <la...@gmail.com> wrote:
>> >
>> > Hi all,
>> >
>> > I am reposting my stackoverflow question here.
>> >
>> > So I would like to do a local test of my pyflink sql application. The sql query does a very simple job: read from source, window aggregation and write result to a sink.
>> >
>> > In the production, the source and sink will be kinesis stream. But since I need to do a local test, I am mocking out the source and sink with the local filesystem.
>> >
>> > The problem is the local job always runs successfully, but the results were never written out to the sink file. Curious what I am missing here.
>> >
>> > Note:
>> >
>> > I am using Flink 1.11.0
>> > if I directly read data from the source table by select query and write to sink without windowing or grouping, it works just fine. That means the source and sink table is set up correctly. So it seems the problem is around the Tumbling and grouping for the local filesystem.
>> > This code works fine with Kinesis source and sink.
>> >
>> >
>> > Here is the SQL code:
>> >
>> > ```
>> >
>> > CREATE TABLE incoming_data (
>> > requestId VARCHAR(4),
>> > groupId VARCHAR(32),
>> > userId VARCHAR(32),
>> > requestStartTime VARCHAR(32),
>> > processTime AS PROCTIME(),
>> > requestTime AS TO_TIMESTAMP(SUBSTR(REPLACE(requestStartTime, 'T', ' '), 0, 23), 'yyyy-MM-dd HH:mm:ss.SSS'),
>> > WATERMARK FOR requestTime AS requestTime - INTERVAL '5' SECOND
>> > ) WITH (
>> > 'connector' = 'filesystem',
>> > 'path' = '/path/to/test/json/file.json',
>> > 'format' = 'json',
>> > 'json.timestamp-format.standard' = 'ISO-8601'
>> > )
>> >
>> > CREATE TABLE user_latest_request (
>> > groupId VARCHAR(32),
>> > userId VARCHAR(32),
>> > latestRequestTime TIMESTAMP
>> > ) WITH (
>> > 'connector' = 'filesystem',
>> > 'path' = '/path/to/sink',
>> > 'format' = 'csv'
>> > )
>> >
>> > INSERT INTO user_latest_request
>> > SELECT groupId,
>> > userId,
>> > MAX(requestTime) as latestRequestTime
>> > FROM incoming_data
>> >
>> > GROUP BY TUMBLE(processTime, INTERVAL '1' SECOND), groupId, userId;
>> >
>> > ```
>> >
>> >
>> > Thanks!
Re: PyFlink SQL window aggregation data not written out to file
Posted by Guoqin Zheng <la...@gmail.com>.
Hi Roman,
Thanks for your quick response!
Yes, it does seem to be the window closing problem.
So if I change the tumble window on eventTime, which is column 'requestTime',
it works fine. I guess the EOF of the test data file kicks in a watermark
of Long.MAX_VALUE.
But my application code needs to calculate the result based on the tumble
window of processTime. Curious what is the typical setup for such a local
test. ie. How do we inform the Flink to close the window if the input
stream hits the end.
Thanks,
-Guoqin
On Mon, Nov 15, 2021 at 11:21 AM Roman Khachatryan <ro...@apache.org> wrote:
> Hi Guoqin,
>
> I think the problem might be related to watermarks and checkpointing:
> - if the file is too small, the only watermark will be the one after
> fully reading the file
> - in exactly once mode, sink waits for a checkpoint completion before
> committing the files
>
> Recently, there were some improvements in handling the last
> checkpoint, so you might want to try Flink 1.14.
> You might also try to configure checkpointing (e.g. decrease interval
> [1]) and watermark generation [2] to diagnose the problem.
>
> [1]
> https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/deployment/config/#execution-checkpointing-interval
> [2]
> https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/sql/create/#watermark
>
>
> Regards,
> Roman
>
> On Mon, Nov 15, 2021 at 6:43 PM Guoqin Zheng <la...@gmail.com>
> wrote:
> >
> > Hi all,
> >
> > I am reposting my stackoverflow question here.
> >
> > So I would like to do a local test of my pyflink sql application. The
> sql query does a very simple job: read from source, window aggregation and
> write result to a sink.
> >
> > In the production, the source and sink will be kinesis stream. But since
> I need to do a local test, I am mocking out the source and sink with the
> local filesystem.
> >
> > The problem is the local job always runs successfully, but the results
> were never written out to the sink file. Curious what I am missing here.
> >
> > Note:
> >
> > I am using Flink 1.11.0
> > if I directly read data from the source table by select query and write
> to sink without windowing or grouping, it works just fine. That means the
> source and sink table is set up correctly. So it seems the problem is
> around the Tumbling and grouping for the local filesystem.
> > This code works fine with Kinesis source and sink.
> >
> >
> > Here is the SQL code:
> >
> > ```
> >
> > CREATE TABLE incoming_data (
> > requestId VARCHAR(4),
> > groupId VARCHAR(32),
> > userId VARCHAR(32),
> > requestStartTime VARCHAR(32),
> > processTime AS PROCTIME(),
> > requestTime AS TO_TIMESTAMP(SUBSTR(REPLACE(requestStartTime,
> 'T', ' '), 0, 23), 'yyyy-MM-dd HH:mm:ss.SSS'),
> > WATERMARK FOR requestTime AS requestTime - INTERVAL '5' SECOND
> > ) WITH (
> > 'connector' = 'filesystem',
> > 'path' = '/path/to/test/json/file.json',
> > 'format' = 'json',
> > 'json.timestamp-format.standard' = 'ISO-8601'
> > )
> >
> > CREATE TABLE user_latest_request (
> > groupId VARCHAR(32),
> > userId VARCHAR(32),
> > latestRequestTime TIMESTAMP
> > ) WITH (
> > 'connector' = 'filesystem',
> > 'path' = '/path/to/sink',
> > 'format' = 'csv'
> > )
> >
> > INSERT INTO user_latest_request
> > SELECT groupId,
> > userId,
> > MAX(requestTime) as latestRequestTime
> > FROM incoming_data
> >
> > GROUP BY TUMBLE(processTime, INTERVAL '1' SECOND), groupId, userId;
> >
> > ```
> >
> >
> > Thanks!
>
Re: PyFlink SQL window aggregation data not written out to file
Posted by Roman Khachatryan <ro...@apache.org>.
Hi Guoqin,
I think the problem might be related to watermarks and checkpointing:
- if the file is too small, the only watermark will be the one after
fully reading the file
- in exactly once mode, sink waits for a checkpoint completion before
committing the files
Recently, there were some improvements in handling the last
checkpoint, so you might want to try Flink 1.14.
You might also try to configure checkpointing (e.g. decrease interval
[1]) and watermark generation [2] to diagnose the problem.
[1] https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/deployment/config/#execution-checkpointing-interval
[2] https://nightlies.apache.org/flink/flink-docs-master/docs/dev/table/sql/create/#watermark
Regards,
Roman
On Mon, Nov 15, 2021 at 6:43 PM Guoqin Zheng <la...@gmail.com> wrote:
>
> Hi all,
>
> I am reposting my stackoverflow question here.
>
> So I would like to do a local test of my pyflink sql application. The sql query does a very simple job: read from source, window aggregation and write result to a sink.
>
> In the production, the source and sink will be kinesis stream. But since I need to do a local test, I am mocking out the source and sink with the local filesystem.
>
> The problem is the local job always runs successfully, but the results were never written out to the sink file. Curious what I am missing here.
>
> Note:
>
> I am using Flink 1.11.0
> if I directly read data from the source table by select query and write to sink without windowing or grouping, it works just fine. That means the source and sink table is set up correctly. So it seems the problem is around the Tumbling and grouping for the local filesystem.
> This code works fine with Kinesis source and sink.
>
>
> Here is the SQL code:
>
> ```
>
> CREATE TABLE incoming_data (
> requestId VARCHAR(4),
> groupId VARCHAR(32),
> userId VARCHAR(32),
> requestStartTime VARCHAR(32),
> processTime AS PROCTIME(),
> requestTime AS TO_TIMESTAMP(SUBSTR(REPLACE(requestStartTime, 'T', ' '), 0, 23), 'yyyy-MM-dd HH:mm:ss.SSS'),
> WATERMARK FOR requestTime AS requestTime - INTERVAL '5' SECOND
> ) WITH (
> 'connector' = 'filesystem',
> 'path' = '/path/to/test/json/file.json',
> 'format' = 'json',
> 'json.timestamp-format.standard' = 'ISO-8601'
> )
>
> CREATE TABLE user_latest_request (
> groupId VARCHAR(32),
> userId VARCHAR(32),
> latestRequestTime TIMESTAMP
> ) WITH (
> 'connector' = 'filesystem',
> 'path' = '/path/to/sink',
> 'format' = 'csv'
> )
>
> INSERT INTO user_latest_request
> SELECT groupId,
> userId,
> MAX(requestTime) as latestRequestTime
> FROM incoming_data
>
> GROUP BY TUMBLE(processTime, INTERVAL '1' SECOND), groupId, userId;
>
> ```
>
>
> Thanks!