zhangjun created FLINK-12584:
--------------------------------
Summary: Add Bucket File Syetem Table Sink
Key: FLINK-12584
URL:
https://issues.apache.org/jira/browse/FLINK-12584 Project: Flink
Issue Type: New Feature
Components: Table SQL / API
Affects Versions: 1.8.0, 1.9.0
Reporter: zhangjun
Assignee: zhangjun
# ** *Motivation*
In flink, the file system (especially hdfs) is a very common output, but for users using sql, it does not support directly using sql to write data to the file system, so I want to add a bucket file system table sink, the user can register it to StreamTableEnvironment, so that table api and sql api can directly use the sink to write stream data to filesystem
# ** *example*
tEnv.connect(new Bucket().basePath("hdfs://localhost/tmp/flink-data"))
.withFormat(new Json().deriveSchema())
.withSchema(new Schema()
.field("name", Types. STRING ())
.field("age", Types. INT ())
.inAppendMode()
.registerTableSink("myhdfssink");
tEnv.sqlUpdate("insert into myhdfssink SELECT * FROM mytablesource");
# ** *Some ideas to achieve this function*
1) Add a class Bucket which extends from ConnectorDescriptor, add some properties, such as basePath.
2) Add a class BucketValidator which extends from the ConnectorDescriptorValidator and is used to check the bucket descriptor.
3) Add a class FileSystemTableSink to implement the StreamTableSink interface. In the emitDataStream method, construct StreamingFileSink for writing data to filesystem according to different properties.
4) Add a factory class FileSystemTableSinkFactory to implement the StreamTableSinkFactory interface for constructing FileSystemTableSink
5) The parameters of withFormat method is the implementation classes of the FormatDescriptor interface, such as Json, Csv, and we can add Parquet、Orc later.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
Locked
