Apache Flink SubmitSql
CertifiedExecute SQL via Flink SQL Gateway
Apache Flink SubmitSql
Execute SQL via Flink SQL Gateway
Runs a SQL statement through the Flink SQL Gateway without uploading a JAR. Respects session settings and waits until the operation reaches a terminal state.
type: io.kestra.plugin.flink.SubmitSqlExamples
Execute a streaming SQL query
id: flink-sql-streaming
namespace: company.team
tasks:
- id: run-sql
type: io.kestra.plugin.flink.SubmitSql
gatewayUrl: "http://flink-sql-gateway:8083"
statement: |
INSERT INTO enriched_orders
SELECT o.order_id, o.customer_id, c.name, o.amount, o.order_time
FROM orders o
JOIN customers c ON o.customer_id = c.id
sessionConfig:
catalog: "default_catalog"
database: "default_database"
configuration:
execution.runtime-mode: "streaming"
execution.checkpointing.interval: "30s"
Execute a batch SQL query
id: flink-sql-batch
namespace: company.team
tasks:
- id: run-batch-sql
type: io.kestra.plugin.flink.SubmitSql
gatewayUrl: "http://flink-sql-gateway:8083"
statement: |
CREATE TABLE daily_summary AS
SELECT DATE(order_time) as order_date,
COUNT(*) as order_count,
SUM(amount) as total_amount
FROM orders
WHERE order_time >= '2024-01-01'
GROUP BY DATE(order_time)
sessionConfig:
configuration:
execution.runtime-mode: "batch"
Properties
gatewayUrl *string
SQL Gateway URL
Base URL of the Flink SQL Gateway (e.g., http://flink-sql-gateway: 8083).
statement *string
SQL statement
SQL statement to execute; supports DDL and DML.
acceptableStates array
Acceptable terminal states
Operation states treated as success. For streaming include RUNNING to keep the session open. Defaults to ['FINISHED', 'RUNNING'].
assets
Assets this task consumes as inputs or produces as outputs, for lineage tracking and the asset graph (Enterprise Edition). A flow declaring this property on a task is rejected in the open-source edition.
io.kestra.core.models.assets.AssetsDeclaration
IGNOREFAILWARNAsset failure behavior
Behavior applied to the task state when a declared asset fails to render, emit, or be persisted (e.g. a lock conflict): FAIL escalates it to FAILED, WARN (default) warns it if it would otherwise succeed, IGNORE leaves the state untouched.
Whether to auto-register assets referenced dynamically at runtime that are not statically declared in inputs or outputs.
The assets consumed as inputs.
io.kestra.core.models.assets.AssetIdentifier
1The assets produced as outputs.
io.kestra.plugin.ee.assets.Dataset
1150{}1150io.kestra.plugin.ee.assets.File
1150{}1150io.kestra.plugin.ee.assets.Table
1150{}1150io.kestra.plugin.ee.assets.VM
1150{}1150io.kestra.core.models.assets.External
1150{}1150io.kestra.core.models.assets.Custom
11501Custom asset type
{}1150connectionTimeout integerstring
30Connection timeout
Connection timeout in seconds; defaults to 30.
sessionConfig string
Session configuration
Session options such as catalog, database, and configuration properties applied before execution.
io.kestra.plugin.flink.SubmitSql-SessionConfig
sessionName string
Session name
Optional session name; a temporary session is created when omitted.
statementTimeout integerstring
300Statement timeout
Execution timeout in seconds for the SQL statement; defaults to 300.
Outputs
operationHandle string
Operation handle
Unique identifier for the executed SQL operation.
resultCount integer
Result count
Number of rows affected or returned; -1 when the gateway does not provide it.
sessionHandle string
Session handle
Identifier for the SQL Gateway session used.
status string
Operation status
Final status reported by the SQL Gateway.
