PluginsPlugin Flink

SubmitSql​Submit​Sql

Submit a SQL statement to Flink SQL Gateway.

This task submits a SQL statement to Apache Flink via the SQL Gateway. No JAR file is required as the SQL is executed directly by Flink.

yaml
type: "io.kestra.plugin.flink.SubmitSql"

Execute a streaming SQL query

yaml
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

yaml
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

SQL Gateway URL

The base URL of the Flink SQL Gateway, e.g., 'http://flink-sql-gateway: 8083'

SQL statement

The SQL statement to execute. Supports both DDL and DML statements.

SubType string

Acceptable terminal states

List of operation states to consider as successful completion. For streaming jobs, include 'RUNNING' - these sessions will be kept alive. For batch jobs, use 'FINISHED'. Defaults to 'FINISHED', 'RUNNING'.

Default 30

Connection timeout

Timeout for connecting to the SQL Gateway in seconds. Defaults to 30.

Session configuration

Session configuration including catalog, database, and Flink configuration properties.

Session name

Optional session name. If not provided, a random session will be created.

Default 300

Statement timeout

Timeout for SQL statement execution in seconds. Defaults to 300.

The operation handle

The unique identifier for the executed SQL operation

Result count

Number of rows affected or returned by the operation

The session handle

The unique identifier for the SQL Gateway session

Operation status

Final status of the operation

SubType string