Questions tagged [flink-sql]

Apache Flink features two relational APIs, SQL and Table API, as unified APIs for stream and batch processing.

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

Column truncated in TableAPI

results displayed by the code below are truncated to 30 : tableEnv.createTemporaryView("topicData", topicData); TableConfig tableConfig = tableEnv.getConfig(); tableConfig.set("table....
Tomasz Krol's user avatar
-1 votes
0 answers
21 views

Apache Flink - ClassNotFoundException

I'm facing few ClassNotFound issues while running Flink jobs. The job is based on Table API and I alos have included the libraries in FlinkShadowJar in my gradle file. java.lang.NoClassDefFoundError: ...
user3497321's user avatar
0 votes
0 answers
17 views

Table result to String

I am trying Dynamic rule with Table API, In my system I have SQL rule coming from Kafka. So I have used a broadcast mechanism to implement the same. But I am stuck on how can I give results back to ...
Keyur Makwana's user avatar
0 votes
0 answers
21 views

Flink SQL: Emitting an event after a certain time has passed after original event was recorded

I am trying to setup a data streaming pipeline that starts with a task to emit an event after a certain time (say 5 days) has passed after a certain other event has been registered in Kafka topic. ...
srdkr13's user avatar
0 votes
1 answer
35 views

Convert JSON schema to Flink DDL

How can I convert a JSON schema to a Flink DDL statement? For example, given the following JSON schema which has a nested object property { "type": "object", "properties&...
slowratatoskr's user avatar
0 votes
0 answers
8 views

In flink, converting from datastream to table performance implications

does convert from datastream to table and back from table to datastream impact performance of an app in a negative way? Tried looking for the flink datastream to table conversion on flink ...
Newbie's user avatar
  • 1
0 votes
0 answers
13 views

Zeppelin with flink and s3 not able to find s3 filesystem

I have flink 1.14.6 installed locally on my machine. In the local flink installation, I have configured flink-s3-fs-hadoop plugin Using sql-client of flink, I am able to work with flink sql and s3 But ...
Yabha's user avatar
  • 21
0 votes
0 answers
15 views

How do I write Flink TIMESTAMP_LTZ to Elasticsearch using ISO-8601?

I have an Elasticsearch SQL sink in a Flink (PyFlink) job, where the sink table looks like: CREATE TABLE mysink ( foo TIMESTAMP_LTZ(3) ) WITH ( ... my elasticsearch connection details ) In my ...
John's user avatar
  • 10.9k
0 votes
0 answers
18 views

How to reference nested objects in Flink cep SQL in MATCH_RECOGNIZE clause

I am using Flink CEP with SQL for processing events, the schema is determined by the JSON event, while trying to execute the given SQL it gives bellow exception can. we have defined DTO’s for the ...
Krutik Gadhiya's user avatar
0 votes
0 answers
29 views

Flink SQL throw "java.lang.ClassNotFoundException: org.apache.hadoop.fs.FSDataInputStream" Error

I want to create a hudi table in my s3 storage,I followed the Official documentation, added these configuration in flink-conf.yml: fs.allowed-fallback-filesystems: s3 state.backend: filesystem state....
Dodge_X's user avatar
  • 37
0 votes
0 answers
14 views

How does the flink operator read external configuration?

I am learning to use Flink to do the pre-aggregation function of metrics. I installed an Agent on a cluster, and it will report many metrics (cpu/memory metrics, and user-defined metrics) every minute,...
xzchaoo's user avatar
  • 51
0 votes
0 answers
21 views

With flink sql with upset-kafka connectot, how to ignore org.apache.kafka.common.errors.RecordTooLargeException

My kafka sink looks like this CREATE TABLE IF NOT EXISTS TABLE_NAME ( org VARCHAR NOT NULL, pod ...
user3822232's user avatar
0 votes
0 answers
32 views

Apache Flink throws exception when submitting a job through Jenkins and Spinnaker

We are using Apache Flink 1.16.1 configured as a standalone Kubernetes pod ,for one of our application to read from confluent Kafka topics to do event correlation. We are using the flink's Table API ...
JDP's user avatar
  • 35
0 votes
0 answers
9 views

Flink cep MATCH_RECOGNIZE not processing all events

I am using MATCH_RECOGNIZE for pattern matching in events, I am trying to find average usage less than 60 withing 10 seconds interval, the issue here is that in result i am only seeing only 2 result (...
Krutik Gadhiya's user avatar
0 votes
1 answer
41 views

Understanding watermark behaviour when using Kafka connector and Flink SQL

I have a problem with understanding how watermarks are assigned when using Kafka connector and Flink SQL. I have inserted messages into kafka topic in given order: 1. A,2023-07-08 09:00:00,2023-07-08,...
Ardelia Lortz's user avatar
0 votes
0 answers
67 views

Cannot load user class: org.apache.flink.connector.kafka.sink.KafkaSink

Flink version: 1.17.1 in local environment through IDE the job is starting and working without any errors, when starting the job through Flink Web ui in ubuntu environment, the job is started ...
Krutik Gadhiya's user avatar
0 votes
1 answer
18 views

In flink how to track a custom metric in a class that extends UserDefinedFunction

I have a simple function that is handling some string which is a json public class ArraySizeUdf extends ScalarFunction { private static final Logger LOG = LoggerFactory.getLogger(ArraySizeUdf....
user3822232's user avatar
0 votes
1 answer
75 views

Apache Flink SQL - Slow Results When Using LAG Function

I have a very simple SQL query, running in AWS Kinesis Data Analytics, reading from a Kinesis Stream. When I use the LAG Function, the performance is really slow, when returning the results. In some ...
Peter Donnelly's user avatar
0 votes
0 answers
55 views

How to Use Map<STRING, OBJECT> dataType (column) in flinkSql

`I am fairly new the flinkSql and for one of our use cases, we need to define a column of type Map<String, Object>. The data is received in json format via and need to materialized and send ...
Rahul S's user avatar
0 votes
0 answers
33 views

Dynamic schema in Table API

I have Flink Job that is using Table API, Input log in the Flink Job has a diff log schema, let’s say log1 has {x=1,y=“abc”,z=true} log2 has {x=1,a=“yu”,b=false} in future I can have other fields As ...
Keyur Makwana's user avatar
0 votes
0 answers
55 views

Flink sql kafka sink with complex data type

I need some help with getting flink kafka connect to produce correct data when data is of complex object type CREATE TABLE IF NOT EXISTS SOURCE_TABLE ( source VARCHAR NOT NULL, id ...
user3822232's user avatar
0 votes
1 answer
105 views

Flink Table API program does not compile when assigning watermark using a field converted with UDF

Since TO_TIMESTAMP(value, format) method in Flink Table API does not support custom formats like yyyyMMddHHmmssSSS, we needed to create a UDF(User Defined Function) for custom conversion. However, ...
Süleyman Fazıl Yeşil's user avatar
0 votes
0 answers
40 views

Aggregation operation on unnest array elements

I have a stream with an array object in it. I want to identify max value among all the elements in the array and forward that max element. Input: { "name": "Dhruv" "...
Dhruv Jagetiya's user avatar
0 votes
0 answers
29 views

Create a UDF function to decode a binary field "payload" based on its schema avsc

I have this schema of payload.avsc { "type": "record", "name": "Data", "fields": [ { "name": "guid", "...
Hsgh775's user avatar
0 votes
0 answers
33 views

Can Hudi source be defined as retract stream in FlinkSQL

I'm trying to use hudi as source to perform aggregation in FlinkSQL,and i found hudi source is append stream but not retract stream.It makes the aggregation results accumulate incorrectly. My codes ...
maple's user avatar
  • 37
0 votes
0 answers
12 views

Does the "accumulate" method in AggregateFunction executes concurrently?

I am trying to aggregate some data using AggregateFunction in flink sql. This is my code: public class SlideWindowHotNumAgg extends AggregateFunction<String, List<Map<String, String>>&...
jingang shi's user avatar
0 votes
1 answer
74 views

Flink SQL Client & Iceberg on MinIo: Class org.apache.hadoop.fs.s3a.S3AFileSystem not found

Running into some issues trying to get a local environment set up to stream data fro Flink into an Iceberg table over MinIO. [ERROR] Could not execute SQL statement. Reason: org.apache.hadoop.hive....
danthelion's user avatar
  • 3,815
0 votes
1 answer
40 views

Using flink upsert-kafka connector is there a way to avoid intermediate messages being written to output kafka topic

I have a kafka source that send event like record_created and record_updated. I perform some transformation and join other kafka topic to this source. Finally emit to a kafka topic. I am using upset-...
user3822232's user avatar
0 votes
0 answers
25 views

Does Flink sql support allowedLateness in sql query now?

How can I apply allowedLateness in this query? select FIRST_VALUE(uid) as uid, mid, count(1) as c from table(TUMBLE(TABLE table_name, DESCRIPTOR(ts), INTERVAL '3' MINUTES)) where target = true group ...
Dax1's user avatar
  • 11
0 votes
1 answer
63 views

Using Flink sql to emit to kafka with buffering producing null pointer exception

I have a sql defined this way CREATE TABLE raw_table ( headers VARCHAR, id VARCHAR, eventTimestamp TIMESTAMP_LTZ(3) NOT NULL, type VARCHAR, contentJson ...
user3822232's user avatar
0 votes
1 answer
21 views

Using flink sql kafka connector running window aggregation

CREATE TABLE server_logs_v1_k ( userid STRING, log_time TIMESTAMP_LTZ(3), WATERMARK FOR log_time AS log_time - INTERVAL '5' SECONDS ) WITH ( 'connector' = 'kafka', 'topic-pattern' = '...
user3822232's user avatar
0 votes
0 answers
13 views

Planner error when calling toDatastream after applying Table Aggregation UDF in Table API

The Table stream is not able to convert to a Datastream, only after calling a User-defined TableAggregateFunction. Using the provided gradle project configuration from Flink and adding the Table API ...
tbone's user avatar
  • 1
0 votes
0 answers
99 views

How to return a array of json using flink sql

I want to return a json that contains an array of pojo. The sql i am using to emit this is INSERT INTO KAFKA_SINK SELECT id, object_arrayagg('id', unnested_segments_id, 'name', ...
user3822232's user avatar
2 votes
0 answers
76 views

Query on Flink SQL primary key for nested field

I have a Kafka topic named employee which uses confluent avro schema and will emit the payload as below: { "employee": { "id": "123456", "name": "...
JDP's user avatar
  • 35
0 votes
0 answers
51 views

Flink WordCount.jar sample file not running on local kubernetes

I have started a local k8s deployment of flink by using their helm chart on my minikube cluster helm install -n flink riskfocus/flink --generate-name After that, on my PC I opened localhost:8081/ ...
Eugenio.Gastelum96's user avatar
0 votes
1 answer
161 views

Using flink sql how to unest a array and join with another table then aggregate the result to json array

I have a table 1 ( source is kafka ) event as { "id": "id-1", "segs": ["seg-id-1", "segs-id-2"] } i have table 2 ( source is kafka) event as { "...
user3822232's user avatar
0 votes
0 answers
34 views

Is there anyway i can use StreamTableEnvrionment in ProcessWindowFunction?

Scenario Using Flink to read MySQL binlog and write to a Hudi table, but I want to partition the binlog data source into windows, and batch insert all the data within a window into the Hudi table when ...
gamblewin's user avatar
0 votes
0 answers
30 views

Output of User-Defined Table Function Cannot to Converted to Stream

I have a datastream that I am converting to a Table so that I may use Flink SQL syntax on the table. My Flink SQL query includes a user-defined Table Function, meaning each input row results in ...
c_mac's user avatar
  • 1
0 votes
0 answers
55 views

Flink SQL - NOT EXISTS query gives error - doesn't support consuming update and delete changes

I have data streaming from kafka on an input topic. Using that data, I have to query another table (either with kafka connector or file system connector) and check if a value does not exist and ...
Neha's user avatar
  • 225
0 votes
1 answer
121 views

Flink SQL automatically uploads Avro schema

I am facing some issues using Flink SQL automatically uploading a schema rather than using the one I have already uploaded. Here is my scenario: I have a Kafka topic "input", which I want to ...
Jonck van der Kogel's user avatar
0 votes
0 answers
32 views

Flink Creating duplicate operators for same processing

I am reading data from Kafka topics with upsert connector and joining around 7 tables. On the joined result performing 3 diffrent aggregation and writing result to kafka sink. flink is creating ...
Vishal Kamlapure's user avatar
0 votes
1 answer
72 views

Decode a field of kafka data from Avro using Flink sql client and Creating a UDF

I want to decode the “load” field from binary to a readable data using a UDF created based on a schema of the avro ( supposing I have the schema data.avsc) Please, Any idea how to create this UDF ...
Hsgh775's user avatar
0 votes
1 answer
40 views

Flink Sql - Row Over Aggregate - Emit strategy

The flink sql docs doesn't share any information about the emit strategy of row over aggregate queries. I guess it has the same behaviour as the windowed aggregation. Only the rows are emitted when a ...
H Sridhar's user avatar
0 votes
0 answers
34 views

Flink Sql Get intermediate results from grouped window aggregation

I know that flink doesn't support out of the box getting intermediate results of the current window of a group aggregation. The results are emitted only for the windows that are closed by the moving ...
H Sridhar's user avatar
0 votes
1 answer
47 views

Flink Sql Delete old state using grouped window aggregation

We are using flink sql to build windowed group aggregation. We are storing the results in the mongo db. We have defined a primary key with table inserting data to mongo db. CREATE VIEW USER_TABLE AS ...
H Sridhar's user avatar
0 votes
0 answers
65 views

Flink SQL lookup table left join dynamic table

I have two tables - Customers and Orders. My requirement is to count the orders of every customer for each minute. I want to use Flink SQL to implement this task. My approach is as follows: CREATE ...
Kerie's user avatar
  • 121
0 votes
0 answers
36 views

Creating a table into a data base on hue from kafka streming data

I want to ceate a table in hue plateform from reading kafka topic but i always get an empty table as shown in the picture : enter image description here this is my example of code using flink sql ...
Hsgh775's user avatar
0 votes
1 answer
116 views

flink sql - extract values from a nested array and concat in one go

I am using flink sql for a very complex object INSERT statement within it there is an array: I already have the array UNNEST using CROSS JOIN in t1...and it is not null (used JSON_STRING on it and got ...
user3017827's user avatar
0 votes
0 answers
40 views

Flink SQL INSERT INTO query results in java.net.ConnectException: Connection refused error

I am trying to run an INSERT INTO query on the Flink SQL Shell but keep running into this error (other queries using commands like CREATE and ALTER work, however). I've tried changing my port and IP ...
compuser016's user avatar
0 votes
0 answers
52 views

State Processor API and Flink SQL jobs

I have a job using a simple SQL query: select file_name, count(*) from source group by file Now using State Processor API to read the checkpoint saved by the job I have to choose proper state object: ...
Ardelia Lortz's user avatar

1
2 3 4 5
14