(DEPRECATED) Apache Flink User Mailing List archive.

How to register TableSinks

Classic

List

Threaded

3 messages Options

Eifler, Patrick

How to register TableSinks

Hi and Happy New Year,

I’m currently trying to remove deprecations to prepare for the upgrade to Flink 1.12. currently running on 1.11.

Specifically I need to update our code that registers table sinks into the StreamTableEnvironment. I’m maintaining jobs that use DataStreams with multiple sinks. Now I want to use the StatementSet to benefit from its DAG for multiple sinks.

So far I added the code to add the sinks into the StatementSet:

statementSet.addInsert(sinkName,.table)

and to execute the StatementSet:

statementSet.execute()

For this to work I need to register the sinks. I used to do that with the (now deprecated) function on the StreamTableEnvironment:

tableEnv.registerTableSink(

sinkName,

fieldNames,

fieldTypes,

tableSink

)

My Question is how to register sinks to be discovered by the statement set? What is the proper replacement for the function registerTableSink?

executeSql(ddl) as suggested, does not apply to this use case. Did not find anything in the documentation either: https://ci.apache.org/projects/flink/flink-docs-stable/dev/table/common.html#translate-and-execute-a-query

When running the job I’m getting the error, that the sink could not be found in the catalog. Which means I have to add the sink into the catalog, but how?

Which function should be used for registering a table sink into the table environments catalog?

Thanks!

Kind Regards,

Patrick

Patrick Eifler

Senior Software Engineer (BI)

Cloud Gaming Engineering & Infrastructure
Sony Interactive Entertainment LLC

Wilhelmstraße 118, 10963 Berlin

Germany

E: [hidden email]

Dawid Wysakowicz-2

Re: How to register TableSinks

Hi Patrick.

Happy New Year to you too ;)

The method you referring was deprecated along with the TableSink whatsoever in favour of a much improved and feature rich new Source & Sink API. You can find an extensive documentation on this new API here[1].

Therefore if you use the old TableSink interface you must stick with the deprecated method.

Best,

Dawid

[1] https://ci.apache.org/projects/flink/flink-docs-master/dev/table/sourceSinks.html

On 04/01/2021 14:53, [hidden email] wrote:

Hi and Happy New Year,

I’m currently trying to remove deprecations to prepare for the upgrade to Flink 1.12. currently running on 1.11.

Specifically I need to update our code that registers table sinks into the StreamTableEnvironment. I’m maintaining jobs that use DataStreams with multiple sinks. Now I want to use the StatementSet to benefit from its DAG for multiple sinks.

So far I added the code to add the sinks into the StatementSet:

statementSet.addInsert(sinkName,.table)

and to execute the StatementSet:

statementSet.execute()

For this to work I need to register the sinks. I used to do that with the (now deprecated) function on the StreamTableEnvironment:

tableEnv.registerTableSink(

        sinkName,

        fieldNames,

        fieldTypes,

        tableSink

)

My Question is how to register sinks to be discovered by the statement set? What is the proper replacement for the function registerTableSink?

executeSql(ddl) as suggested, does not apply to this use case. Did not find anything in the documentation either: https://ci.apache.org/projects/flink/flink-docs-stable/dev/table/common.html#translate-and-execute-a-query

When running the job I’m getting the error, that the sink could not be found in the catalog. Which means I have to add the sink into the catalog, but how?

Which function should be used for registering a table sink into the table environments catalog?

Thanks!

Kind Regards,

Patrick

--

Patrick Eifler

Senior Software Engineer (BI)

Cloud Gaming Engineering & Infrastructure
Sony Interactive Entertainment LLC

Wilhelmstraße 118, 10963 Berlin

Germany

E: [hidden email]

signature.asc (849 bytes) Download Attachment