You are viewing a plain text version of this content. The canonical link for it is here.
Posted to user@flink.apache.org by Pa...@sony.com on 2021/01/04 13:53:03 UTC

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: patrick.eifler@sony.com

Re: How to register TableSinks

Posted by Pa...@sony.com.
Hey, Thanks Dawid,

One more question: Does the StatementSet API supposed to work with the old sink interface?
I get the following error when I’m using it with the deprecated registerTableSink method:

The main method caused an error: requirement failed: operations should not be empty

Thanks!

Patrick
--
Patrick Eifler

Senior Software Engineer (BI)

Cloud Gaming Engineering & Infrastructure
Sony Interactive Entertainment LLC

Wilhelmstraße 118, 10963 Berlin

Germany

E: patrick.eifler@sony.com

From: Dawid Wysakowicz <dw...@apache.org>
Date: Monday, 4. January 2021 at 15:50
To: <Pa...@sony.com>, <us...@flink.apache.org>
Subject: 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, Patrick.Eifler@sony.com<ma...@sony.com> 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: patrick.eifler@sony.com<ma...@sony.com>

Re: How to register TableSinks

Posted by Dawid Wysakowicz <dw...@apache.org>.
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, Patrick.Eifler@sony.com 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: patrick.eifler@sony.com
>