You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by Yik San Chan <ev...@gmail.com> on 2021/04/21 11:44:14 UTC

Flink: Not able to sink a stream into csv

The question is cross posted on Stack Overflow
https://stackoverflow.com/questions/67195207/flink-not-able-to-sink-a-stream-into-csv
.

I am trying to sink a stream into filesystem in csv format using PyFlink,
however it does not work.

```python
# stream_to_csv.py
from pyflink.table import EnvironmentSettings, StreamTableEnvironment

env_settings =
EnvironmentSettings.new_instance().in_streaming_mode().use_blink_planner().build()
table_env = StreamTableEnvironment.create(environment_settings=env_settings)

table_env.execute_sql("""
    CREATE TABLE datagen (
        id INT,
        data STRING
    ) WITH (
        'connector' = 'datagen',
        'rows-per-second' = '1'
    )
""")

table_env.execute_sql("""
    CREATE TABLE print (
        id INT,
        data STRING
    ) WITH (
        'connector' = 'filesystem',
        'format' = 'csv',
        'path' = '/tmp/output'
    )
""")

table_env.execute_sql("""
INSERT INTO print
SELECT id, data
FROM datagen
""").wait()
```

To run the script:

```
$ python stream_to_csv.py
```

I expect records go to /tmp/output folder, however that doesn't happen.

```
$ ~ ls /tmp/output
(nothing shown here)
```

Anything I miss?

Best,
Yik San

Re: Flink: Not able to sink a stream into csv

Posted by Yik San Chan <ev...@gmail.com>.
Hi Dian,

Thanks for your help, again!

Best,
Yik San

On Wed, Apr 21, 2021 at 8:39 PM Dian Fu <di...@gmail.com> wrote:

> Hi Yik San,
>
> You need to set the rolling policy for filesystem. You could refer to the
> Rolling Policy section [1] for more details.
>
> Actually there are output and you could execute command `ls -la
> /tmp/output/`, then you will see several files named “.part-xxx”.
>
> For your job, you need to set the `execution.checkpointing.interval` in
> the configuration and `sink.rolling-policy.rollover-interval` in the
> property of Filesystem connector.
>
>
> The job will look like the following:
> ```
> from pyflink.table import EnvironmentSettings, StreamTableEnvironment
>
> env_settings =
> EnvironmentSettings.new_instance().in_streaming_mode().use_blink_planner().build()
> table_env =
> StreamTableEnvironment.create(environment_settings=env_settings)
> table_env.get_config().get_configuration().set_string("execution.checkpointing.interval",
> "10s")
>
> table_env.execute_sql("""
>     CREATE TABLE datagen (
>         id INT,
>         data STRING
>     ) WITH (
>         'connector' = 'datagen',
>         'rows-per-second' = '1'
>     )
> """)
>
> table_env.execute_sql("""
>     CREATE TABLE print (
>         id INT,
>         data STRING
>     ) WITH (
>         'connector' = 'filesystem',
>         'format' = 'csv',
>         'path' = '/tmp/output',
>         'sink.rolling-policy.rollover-interval' = '10s'
>     )
> """)
>
> table_env.execute_sql("""
> INSERT INTO print
> SELECT id, data
> FROM datagen
> """).wait()
> ```
>
> [1]
> https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/table/connectors/filesystem.html#rolling-policy
>
> 2021年4月21日 下午7:44，Yik San Chan <ev...@gmail.com> 写道：
>
> The question is cross posted on Stack Overflow
> https://stackoverflow.com/questions/67195207/flink-not-able-to-sink-a-stream-into-csv
> .
>
> I am trying to sink a stream into filesystem in csv format using PyFlink,
> however it does not work.
>
> ```python
> # stream_to_csv.py
> from pyflink.table import EnvironmentSettings, StreamTableEnvironment
>
> env_settings =
> EnvironmentSettings.new_instance().in_streaming_mode().use_blink_planner().build()
> table_env =
> StreamTableEnvironment.create(environment_settings=env_settings)
>
> table_env.execute_sql("""
>     CREATE TABLE datagen (
>         id INT,
>         data STRING
>     ) WITH (
>         'connector' = 'datagen',
>         'rows-per-second' = '1'
>     )
> """)
>
> table_env.execute_sql("""
>     CREATE TABLE print (
>         id INT,
>         data STRING
>     ) WITH (
>         'connector' = 'filesystem',
>         'format' = 'csv',
>         'path' = '/tmp/output'
>     )
> """)
>
> table_env.execute_sql("""
> INSERT INTO print
> SELECT id, data
> FROM datagen
> """).wait()
> ```
>
> To run the script:
>
> ```
> $ python stream_to_csv.py
> ```
>
> I expect records go to /tmp/output folder, however that doesn't happen.
>
> ```
> $ ~ ls /tmp/output
> (nothing shown here)
> ```
>
> Anything I miss?
>
> Best,
> Yik San
>
>
>

Re: Flink: Not able to sink a stream into csv

Posted by Dian Fu <di...@gmail.com>.
Hi Yik San,

You need to set the rolling policy for filesystem. You could refer to the Rolling Policy section [1] for more details.

Actually there are output and you could execute command `ls -la /tmp/output/`, then you will see several files named “.part-xxx”.

For your job, you need to set the `execution.checkpointing.interval` in the configuration and `sink.rolling-policy.rollover-interval` in the property of Filesystem connector.


The job will look like the following:
```
from pyflink.table import EnvironmentSettings, StreamTableEnvironment

env_settings = EnvironmentSettings.new_instance().in_streaming_mode().use_blink_planner().build()
table_env = StreamTableEnvironment.create(environment_settings=env_settings)
table_env.get_config().get_configuration().set_string("execution.checkpointing.interval", "10s")

table_env.execute_sql("""
    CREATE TABLE datagen (
        id INT,
        data STRING
    ) WITH (
        'connector' = 'datagen',
        'rows-per-second' = '1'
    )
""")

table_env.execute_sql("""
    CREATE TABLE print (
        id INT,
        data STRING
    ) WITH (
        'connector' = 'filesystem',
        'format' = 'csv',
        'path' = '/tmp/output',
        'sink.rolling-policy.rollover-interval' = '10s'
    )
""")

table_env.execute_sql("""
INSERT INTO print
SELECT id, data
FROM datagen
""").wait()
```

[1] https://ci.apache.org/projects/flink/flink-docs-release-1.12/dev/table/connectors/filesystem.html#rolling-policy

> 2021年4月21日 下午7:44，Yik San Chan <ev...@gmail.com> 写道：
> 
> The question is cross posted on Stack Overflow https://stackoverflow.com/questions/67195207/flink-not-able-to-sink-a-stream-into-csv <https://stackoverflow.com/questions/67195207/flink-not-able-to-sink-a-stream-into-csv>.
> 
> I am trying to sink a stream into filesystem in csv format using PyFlink, however it does not work.
> 
> ```python
> # stream_to_csv.py
> from pyflink.table import EnvironmentSettings, StreamTableEnvironment
> 
> env_settings = EnvironmentSettings.new_instance().in_streaming_mode().use_blink_planner().build()
> table_env = StreamTableEnvironment.create(environment_settings=env_settings)
> 
> table_env.execute_sql("""
>     CREATE TABLE datagen (
>         id INT,
>         data STRING
>     ) WITH (
>         'connector' = 'datagen',
>         'rows-per-second' = '1'
>     )
> """)
> 
> table_env.execute_sql("""
>     CREATE TABLE print (
>         id INT,
>         data STRING
>     ) WITH (
>         'connector' = 'filesystem',
>         'format' = 'csv',
>         'path' = '/tmp/output'
>     )
> """)
> 
> table_env.execute_sql("""
> INSERT INTO print
> SELECT id, data
> FROM datagen
> """).wait()
> ```
> 
> To run the script:
> 
> ```
> $ python stream_to_csv.py
> ```
> 
> I expect records go to /tmp/output folder, however that doesn't happen.
> 
> ```
> $ ~ ls /tmp/output
> (nothing shown here)
> ```
> 
> Anything I miss?
> 
> Best,
> Yik San