Questions tagged [flink-sql]
Apache Flink features two relational APIs, SQL and Table API, as unified APIs for stream and batch processing.
669
questions
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....
-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: ...
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 ...
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.
...
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&...
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 ...
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 ...
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 ...
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 ...
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....
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,...
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 ...
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 ...
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 (...
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,...
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 ...
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....
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 ...
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 ...
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 ...
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 ...
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, ...
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"
"...
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",
"...
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 ...
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>>&...
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....
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-...
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 ...
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 ...
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' = '...
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 ...
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', ...
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": "...
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/ ...
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
{
"...
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 ...
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 ...
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 ...
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 ...
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 ...
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 ...
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 ...
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 ...
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
...
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 ...
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 ...
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 ...
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 ...
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: ...