SQL Node

The SQL node is a powerful transformation component in ZephFlow that allows you to apply SQL queries to transform, filter, or enrich incoming data. It uses Fleak SQL, a PostgreSQL-compatible dialect with enhanced functionality for handling nested data structures and JSON operations.

Key Features

• Transform data using familiar SQL syntax
• Filter and aggregate records with standard SQL queries
• Extract and manipulate nested JSON data with powerful operators
• Apply complex transformations with SQL functions
• Process data in a declarative way

Attaching SQL Node in a DAG

jobContext:
  otherProperties: {}
  metricTags: {}

dag:
  - id: "sql_node"
    commandName: "sqleval"
    config:
      sql: "your SQL query"
    outputs:
      - "next_node"

Basic Usage

Simple Transformation Example

SELECT
  id,
  timestamp,
  data->>'message' AS message,
  status
FROM records
WHERE status = 'success'

This SQL query:

• Selects specific fields from the incoming records
• Extracts the 'message' field from a JSON 'data' object
• Filters to include only records with 'success' status

JSON Transformation Example

SELECT
  id,
  json_build_object(
    'user', data->'user',
    'transaction_id', transaction_id
  ) AS processed_data
FROM records
WHERE data->'user'->>'role' = 'admin'

This transforms each record by:

• Creating a new JSON structure with selected fields
• Including the entire 'user' object from the JSON 'data' field
• Filtering to include only records with user role 'admin'

Config Object

Field	Type	Required	Default	Description
sql	String	Yes	—	SQL query to execute against the incoming records. Uses Fleak SQL, a PostgreSQL-compatible dialect

Examples

Field Selection and Renaming

SELECT id AS event_id, timestamp AS event_time, message AS content FROM records

Working with JSON Data

SELECT id, data->'user'->>'name' AS username, data->'metrics'->>'response_time'::numeric AS response_time FROM records

Filtering with Complex Conditions

SELECT * FROM records WHERE status = 'error' AND data->>'severity'::int > 3 AND ts > '2023-01-01'

Aggregation and Grouping

SELECT status, COUNT(*) AS count, SUM(response_time::numeric) AS total_response_time FROM records GROUP BY status

Log Parsing with Grok Function

SELECT *, grok('%{TIMESTAMP_ISO8601:timestamp} %{LOGLEVEL:level} %{GREEDYDATA:message}', raw_log) AS parsed FROM records

Conditional Expressions with CASE

SELECT id, CASE WHEN response_time < 100 THEN 'fast' WHEN response_time < 300 THEN 'normal' ELSE 'slow' END AS performance FROM records

SQL Node vs. Eval Node

The SQL node and Eval node both provide data transformation capabilities, but with different syntax and capabilities:

Feature	SQL Node	Eval Node
Syntax	SQL-based (declarative)	FEEL-based (expression language)
Output Structure	Defined by SELECT clause	Must return a dictionary/map
Selection	Explicit column selection	Maps input to output structure
Filtering	WHERE clause	Requires separate filter node
Aggregation	Built-in (GROUP BY)	Requires custom implementation
JSON Handling	-> and ->> operators	Direct path expressions ($.field)
Ideal For	Relational transformations, aggregations	Custom object creation, complex transformations

Best Practices

1. Be Mindful of Column Names

Column names in the SELECT clause define the structure of the output records. Use aliases (AS) to create meaningful output field names.

-- Good: Clear field names
SELECT timestamp AS event_time, message AS event_message FROM records

-- Avoid: Implicit column names from expressions
SELECT timestamp, data->>'message' FROM records

2. Use JSON Operators Correctly

• Use -> to access nested JSON fields as JSON objects
• Use ->> only for the final extraction as text
• Cast types when needed for numeric operations

-- Good: Proper JSON navigation and type casting
SELECT data->'user'->'address'->>'zip_code'::int AS zip FROM records

-- Avoid: Using ->> in the middle of a chain
SELECT data->>'user'->>'address'->>'zip_code' AS zip FROM records

3. Optimize Queries

• Select only the fields you need
• Use appropriate filters to reduce the data volume

Related Nodes

• eval: Transform data using the Fleak Eval Expression Language (FEEL)
• filter: Filter data based on specific conditions using FEEL
• assertion: Validate that data meets specific conditions using FEEL

Limitations

• The SQL node can only access data that is part of the input records
• For Kafka sources with single-message inputs, aggregation functions operate only on the current message
• Table operations are limited since all data comes from the virtual records table

For more details about the SQL language supported by this node, please refer to the Fleak SQL Reference.