Hi.
I am trying to use the FileSystem
<https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/connectors/table/filesystem>
connector
to simply read data from a text file and then write that data to an output
CSV file.
I notice that Flink allows specifying the name of the output file by
setting the sink.partition-commit.success-file.name
<https://ci.apache.org/projects/flink/flink-docs-release-1.13/docs/connectors/table/filesystem/#sink-partition-commit-success-file-name>
option
when defining the schema as mentioned in the docs.
I have configured 'sink.partition-commit.success-file.name' = 'output.csv' in
my output table. The content of the output file is as expected, but it
still comes with a long and complex name like
part-f6c22645-2f03-4062-a22c-8892743d5626-3-0.
What am I missing?
import os
from pyflink.table import TableEnvironment, EnvironmentSettings
settings = EnvironmentSettings.new_instance(
).in_streaming_mode().use_blink_planner().build()
table_env = TableEnvironment.create(settings)
def set_up_tables():
input_file_path = os.path.join(os.path.abspath(os.path.dirname(__file__)),
'../data/input.txt')
output_dir_path = os.path.join(os.path.abspath(os.path.dirname(__file__)),
'../data')
table_env.execute_sql("""
CREATE TABLE input(
`value` BINARY
) WITH (
'connector' = 'filesystem',
'path' = 'file://{}',
'format' = 'raw'
)
""".format(input_file_path))
table_env.execute_sql("""
CREATE TABLE output(
`result` BINARY
) WITH (
'connector' = 'filesystem',
'path' = 'file://{}',
'format' = 'csv',
'sink.partition-commit.policy.kind' = 'success-file',
'sink.partition-commit.success-file.name' = 'output.csv'
)
""".format(output_dir_path))
input = table_env.from_path('input')
output = table_env.from_path('output')
input.print_schema()
output.print_schema()
def demo():
set_up_tables()
input = table_env.from_path('input')
input.execute_insert('output').wait()
if __name__ == '__main__':
demo()
--
------------------------------------------------------------
--------------------------------------------------
Nguyen Dich Long,
School of Information and Communication Technology (SoICT -
https://www.soict.hust.edu.vn)
Hanoi University of Science and Technology (https://www.hust.edu.vn)
601, B1 Building - No 1, Dai Co Viet Street, Hai Ba Trung District, Ha Noi,
Vietnam
Tel: +84 (0)3.54.41.76.76
Email: long.nd162...@sis.hust.edu.vn; longnguyen25111...@gmail.com
