DataWorks:Flink SQL streaming node

Prerequisites

Before you begin, make sure you have:

• Attached a Realtime Compute for Apache Flink computing resource in the Management Center. For more information, see Attach a computing resource.

• Created a Flink SQL Streaming node. For more information, see Create a node for a scheduled workflow.

Step 1: Write the SQL code

In the SQL editing area on the Flink SQL Streaming node editing page, write your task code. To pass values dynamically at runtime, define variables using the ${variable_name} format in your SQL, then assign values in the Script Parameters section of the Real-time Configuration pane on the right.

The following example creates a source table and a sink table, then filters rows based on a parameterized name length:

-- Create the source table datagen_source.
CREATE TEMPORARY TABLE datagen_source(
  name VARCHAR
) WITH (
  'connector' = 'datagen'
);

-- Create the sink table blackhole_sink.
CREATE TEMPORARY TABLE blackhole_sink(
  name VARCHAR
) WITH (
  'connector' = 'blackhole'
);

-- Insert data from the source table into the sink table.
INSERT INTO blackhole_sink
SELECT
  name
FROM datagen_source WHERE LENGTH(name) > ${name_length};

In this example, set name_length to 5 in Script Parameters. The job then filters out rows where the name is 5 characters or shorter.

Step 2: Configure the node

Step 3: Publish and start the node

What's next

• Configure job resources — fine-tune JobManager and TaskManager allocation for production workloads

• Configure job deployment information — full reference for Flink runtime parameters

• Publish a node or workflow — publish workflows that include multiple nodes