Questions tagged [flink-table-api]

Apache Flink is an open source platform for scalable batch and stream data processing. Flink supports batch and streaming analytics, in one system. Analytical programs can be written in concise and elegant APIs in Java and Scala. Apache Flink features Table API (and SQL API) as unified APIs for stream and batch processing.

Filter by
Sorted by
Tagged with
0 votes
1 answer
54 views

Consuming json message

I created a flink application using the table API to ingest data from a kafka topic (that I generate myself). The dataset is youtube stats from kaggle. I can see in confluent's UI that the topic is ...
1 vote
0 answers
62 views

Flink Table API : Flink Dynamic table produces incorrect intermediate CDC values

I am using flink table API to calculate few aggregations. I have stream of data coming from Kafka which is transformed to stream of rows. Using this rows I am creating dynamic table. Ex: consider ...
1 vote
0 answers
15 views

Is it currently possible to generate a changelog steam with the Data Generator Source from FLIP-238

In FLIP-238 and the related merge, new DataGeneratorSource was introduced. It appears that this allows us to easily create new DataGeneration sources. However it's not clear if in it's current form ...
  • 59
0 votes
0 answers
34 views

Lookup join or enrichment join against a changelog stream in Apache Flink Table API

I'm interested in doing a "lookup join" or "enrichment join" against a "changelog stream" read by "upsert-kafka". I am wondering if this is possible against the ...
  • 59
0 votes
1 answer
134 views

Error in Implementing Flink SQL Processing Time Temporal Left Join

I have a stream of data coming from Kafka which I want to enrich with a static data stored in Parquet files in Hadoop, and finally write to a Filesystem sink. Initially I tried a lookup join as below, ...
  • 1
0 votes
0 answers
25 views

Flink: Wait for Kafka CDC to playback from earliest-offset at start up

I am trying to implement a way to wait until the table kafka-connector has caught up with the latest offset. I've already tried implementing a session gap window operator on that table (kafka-...
0 votes
1 answer
55 views

Flink: Temporal Join not emitting data

I'm trying to implement a event-time temporal join but I don't see any data being emitted from the join. I don't see any runtime exceptions either. Flink Version: 1.13 Kafka topics have only 1 ...
0 votes
1 answer
210 views

How to set not null column via Schema.Builder in Flink?

When I was create table via Table API, my code looks like this: Schema.Builder schemaBuilder = Schema.newBuilder(); schemaBuilder.column("id", DataTypes.BIGINT()) .column("value&...
  • 1
0 votes
1 answer
92 views

Flink TableEnvironment.create throws NoSuchMethodError

I am testing flink hive connector, following the instruction here https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/connectors/table/hive/overview/. The final code is as follows. I tried ...
0 votes
1 answer
78 views

what state/s does flink create when using the tableApi to join multiple tables?

I have multiple streams that I want to join (A to B, B to C , C to D...) to create one Z when using the table api and joining 3 tables select * from A inner join B on a.pk_id = b.fk_id inner join C ...
  • 3
0 votes
0 answers
221 views

Flink SQL : How to unpack fields in ROW type as multiple columns?

I call a UDF in such a Flink SQL query: SELECT dvid, rank_name, rank_type, window_start, window_end, RankDif(rank_order,rank_pt) AS rank_cur FROM TABLE( HOP(TABLE UniqueRankTable, DESCRIPTOR(...
0 votes
1 answer
211 views

Headers with Flink Table Api Kafka Sink

How does one send kafka headers with the table api in Flink? I used the sql syntax to create a sink but not sure how to include custom headers.
  • 416
0 votes
1 answer
36 views

Read a csv file in Flink 1.15 using Table API

I'm doing a simple tutorial with flink + java using Table API. What I want to do is really simple - I want to read a csv file from a local filesystem, using a schema and print it out. The way I'm ...
  • 2,221
0 votes
1 answer
68 views

Is there a Flink Table API equivalent to Window Functions using row_number(), rank(), dense_rank()?

In an attempt to discover the possibilities and limitations of the Flink Table API for use in a current project, I was trying to translate a Flink SQL statement into its equivalent Flink Table API ...
0 votes
1 answer
111 views

Flink table left outer join showing an error

I am doing a left outer join on two tables in flink, and the code is given below, showing an exception, inner join with the same two tables worked fine and was able to convert to Datastream Table ...
1 vote
1 answer
242 views

"Cannot map checkpoint/savepoint state for operator" when using fromChangelogStream

I want to use the savepoint mechanism to move existing jobs from one version of Flink to another, by: Stopping a job with a savepoint Creating a new job from the savepoint, on the new version. Until ...
1 vote
0 answers
177 views

In Flink table API, how do you use postgres timestamps in scan.partition.column scan.partition.lower-bound etc

In Flink 1.13, how do you configure a CREATE TABLE statement to use a postgres timestamp column to partition by? Things I have tried: In postgres, I have a column named 'my_timestamp' of type ...
0 votes
1 answer
43 views

java return data type of flink table api aggregation function

I am using flink table api to compute avg and stddevPop for some fields as the following: Table windowedTable = inputTable .window(Tumble.over(lit(5).seconds()).on($("ts"))....
0 votes
0 answers
213 views

Empty field are illegal when writing parquet

I'm trying to write parquet files with table api in flink. I have a nested schema and even tough there are not any empty fields I get a 'empty fields are illegal, the field should be ommited ...
0 votes
0 answers
180 views

Integrate flink environment with hive

Hi I am trying to integrate my flink environment with hive. I am using flink 1.15.1 and hive 1.1 I read in the flink doc here that I need to add to the /lib directory the flink-sql-connector-hive-1.2....
0 votes
1 answer
205 views

Flink table api filesystem connector is not available

I'm trying to write to my local filesystem with flink table api (1.15.1 version). I'm using TableDescriptor.forConnector('filesystem') but I get the exception: Could not find any factory for ...
0 votes
0 answers
55 views

Flink table api writes empty parquet

I try to write parquet files with Flink table api but the files that are written empty. I use StreamExecutionTable and write the files to my local filesystem. When I write jsons the files are written ...
0 votes
1 answer
146 views

Apache Flink Table API Insert statement

I have a Flink application that processes data from 2 streams. I am using a Table API where i want to consume data from one stream1 and query another stream2 and get the record with the latest ...
  • 79
1 vote
0 answers
81 views

Unexpected type: BINARY

I am trying to read parquet files via the Flink table, and it throws the error when I select one of the timestamps. My parquet table is something like this. I create a table with this SQL : ...
  • 320
1 vote
1 answer
390 views

java.lang.NoClassDefFoundError: org/apache/flink/table/types/logical/LogicalTypeRoot

I am trying to read parquet files via the ParquetAvroInputFormat but it throws NoClassDefFoundError exceptions. I find the class in the flink-table-common library. It is in the path org.apache.flink....
  • 320
0 votes
1 answer
271 views

Flink Table API streaming S3 sink throws SerializedThrowable exception

I am trying to write a simple table API S3 streaming sink (csv format) using flink 1.15.1 and I am facing the following exception, Caused by: org.apache.flink.util.SerializedThrowable: ...
0 votes
1 answer
190 views

Duplicate messages in Flink kafka sink from Source kafka CDC using table connectors

I'm trying to use Apache Flink 1.15.1 to read messages from two kafka topic, transform them and finally send them to another kafka topic. Added some savepoints to save the state of the application in ...
0 votes
1 answer
79 views

flink table API not processing records

I read json data from Kafka and tried to process the data with flink table API. StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); StreamTableEnvironment tEnv = ...
0 votes
0 answers
634 views

org.apache.flink.table.api.TableException: Failed to execute sql

I have a problem running a flink job in mini kube that use table connectors to read messages from kafka and then tries to write back to kafka (am just trying to learn table api so not doing any ...
0 votes
0 answers
28 views

Can't insert records into a kinesis queue from flink?

I have an Apache Flink process that reads from a kinesis stream, makes some queries and I want the result to be dumped to another kinesis stream (version 1.13) The whole process is Ok, except for the ...
0 votes
1 answer
113 views

Java Flink 1.13 Exception trying to use tableAPI

This is the error I'm getting trying to initialize StreamTableEnvironment private final StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env); Exception in thread "main" java....
0 votes
0 answers
268 views

Datatype extraction not working when registering UDF with .createTemporaryFunction()

I have a table "vertices" with a custom datatype "Properties", which implements a HashMap<String, PropertyValue> and is interpreted by Flink as RAW('org...impl.properties....
  • 11
0 votes
1 answer
226 views

Flink Table print connector not being called

I am using the Flink table API to pull data from a kinesis topic into a table. I want to periodically pull that data into a temporary table and run a custom scalar function on it. However, I notice ...
0 votes
1 answer
518 views

Apache Flink Create Table

I'm trying to create table in flink using Table API in Java using eclipse using the following code. EnvironmentSettings settings = EnvironmentSettings .newInstance() ....
  • 3
0 votes
1 answer
648 views

Flink failed to trigger checkpoint when using table API

My flink streaming application (v1.14.4) contain JDBC connector used for initial fetch data from MySQL server Logic: JDBC table source -> select.where() -> convert to datastream Kafka ...
1 vote
1 answer
158 views

How can I join two continuous queries in Flink Table API?

I'd like to stack up two continuous queries (views based on a single upstream connector) together and eventually be consistent in my sink result at the end of the pipeline. The first view will remove ...
  • 361
0 votes
1 answer
706 views

Why Flink Table SQL API upsert-kafka sink connector doesn't create a log compacted topic?

I'm trying to replicate Flink's upsert-kafka connector example. Using the following input: event_id,user_id,page_id,user_region,viewtime e0,1,11,TR,2022-01-01T13:26:41.298Z e1,1,22,TR,2022-01-02T13:26:...
  • 361
0 votes
2 answers
420 views

How to create a DataStreamSource from a Mysql Database?

I have a problem running a flink job that is basically running a query against a mysql database and then tries to create a temporary view that must be accessed from a different job. public static void ...
  • 145
1 vote
1 answer
721 views

Difference between Flink mysql and mysql-cdc connector?

In order to enrich the data stream, we are planning to connect the MySQL (MemSQL) server to our existing flink streaming application As we can see that Flink provides a Table API with JDBC connector ...
5 votes
1 answer
899 views

How to convert a Table to a DataStream containing array types (Flink)?

I have issues concerning the table-api of Flink (1.13+). I have a POJO containing several fields, one of them being: List<String> my_list; I create my table using the following declaration for ...
  • 163
0 votes
1 answer
361 views

Flink fails to load class from JAR added via PipelineOptions

I am developing a Java application which uses UDFs on Flink 1.14. I am using PipelineOptions.JARS config to add jar files containing UDF classes dynamically in the application code, However ...
  • 155
3 votes
2 answers
699 views

Apache Flink 1.14.0 - Unable to use python UDF through SQL DDL in Java

I'm trying to execute the python UDF function in SQL DDL(1.14.0) Python file here: from pyflink.table import DataTypes from pyflink.table.udf import udf @udf(input_types=[DataTypes.INT()], ...
  • 196
0 votes
0 answers
153 views

no results when applying flink tumble window in flink table api

I have the same problem when I study https://nightlies.apache.org/flink/flink-docs-release-1.14/zh/docs/try-flink/table_api/. When I use tumble window, I got nothing in MySQL sink. return transactions ...
  • 381
0 votes
2 answers
463 views

No results in kafka topic sink when applying tumble window aggregation in Flink Table API

I am using Flink 1.14 deployed by lyft flink operator I am trying to make tumble window aggregate with the Table API, read from the transactions table source, and put the aggregate result by window ...
1 vote
1 answer
403 views

how to get the last result when doing table join (using toRetractStream in flink sql

Here is my code using flink sql API to join two tables tEnv.createTemporaryView("A", streamA,"speed_sum,cnt,window_start_time,window_end_time"); tEnv.createTemporaryView("B&...
  • 31
0 votes
2 answers
673 views

Could not find any factory for identifier 'avro-confluent' that implements 'org.apache.flink.table.factories.DeserializationFormatFactory'

I have a Flink job that runs well locally but fails when I try to flink run the job on cluster. The error happens when trying to load data from Kafka via 'connector' = 'kafka'. I am using Flink-Table ...
3 votes
2 answers
338 views

Flink sink filesystem as parquet - error on saving nested data

I am trying to convert a json data to parquet so than I can use Trino or presto to query. Sample JSON is as follows: {"name": "success","message": "test","...
2 votes
0 answers
444 views

Flink Sql not converting RAW('org.apache.avro.util.Utf8', '...') to String

I'm reading from a kafka stream, creating a Table environment and calculating an average and writing the data back to kafka [SIMPLECUSTOMER]. This worked in Flink 1.12.5. I'm using Flink 1.13.2 and ...
0 votes
1 answer
743 views

How to map java LocalDateTime to Flink TIMESTAMP when using table API

My code is something like: DataStreamSource<Tuple2<String, LocalDateTime>> src = ...; tableEnv.createTemporaryView("input_table", src, $("name"), $("dt")); I ...
  • 1,677
0 votes
0 answers
419 views

Apache Flink version 1.13 Convert Table to Dataset?

I am converting some legacy Java code written for Flink version 1.5 to Flink version 1.13.1. Specifically, I'm working with Table API. I have to read data from CSV file, perform some basic SQL and ...
  • 294