How to reference nested JSON within PyFlink SQL when JSON schema varies?

Question

I have a stream of events I wish to process using PyFlink, where the events are taken from AWS EventBridge. The events in this stream share a number of common fields, but their detail field varies according to the value of the source and/or detail-type field. For example, here is an example event from EC2:

{
  "version": "0",
  "id": "6a7e8feb-b491-4cf7-a9f1-bf3703467718",
  "detail-type": "EC2 Instance State-change Notification",
  "source": "aws.ec2",
  "account": "111122223333",
  "time": "2017-12-22T18:43:48Z",
  "region": "us-west-1",
  "detail": {
    "instance-id": " i-1234567890abcdef0",
    "state": "terminated"
  }
}

The id, version, source etc fields are consistent across event types, but note that a different type of event in the stream would have a different shape to the detail field, e.g. a CodeBuild event might look like this:

  "detail":{
    "build-status": "SUCCEEDED",
    "project-name": "my-sample-project",
    "build-id": "arn:aws:codebuild:us-west-2:123456789012:build/my-sample-project:8745a7a9-c340-456a-9166-edf953571bEX",
    "additional-information": {
      "artifact": {
        "md5sum": "da9c44c8a9a3cd4b443126e823168fEX",
        "sha256sum": "6ccc2ae1df9d155ba83c597051611c42d60e09c6329dcb14a312cecc0a8e39EX",
        "location": "arn:aws:s3:::codebuild-123456789012-output-bucket/my-output-artifact.zip"
      }
     }
   }

I'd like to create a statement that looks like the below to key the stream by detail-type to feed different sub-tables, where each sub-table is processed differently.

INSERT INTO ec2_event_table SELECT * from input_table WHERE source = 'aws.ec2'
INSERT INTO codebuild_event_table SELECT * from input_table WHERE source = 'aws.codebuild'

How do I define input_table (the input table containing the multiplexed events)? I've tried:

CREATE TABLE input_table (
            source VARCHAR,
            detail MAP
          )

but this gives me an error. I need to specify the type of MAP, e.g. MAP<VARCHAR, VARCHAR>, I can't use MAP<>.

How do I refer to deeply nested JSON using PyFlink SQL?

Is what I am trying to do possible with SQL and the Table API, or do I need to use the DataStream API? I don't want to create a different input stream for each different event type.

score 3 · Accepted Answer · answered Apr 27 '22 at 01:54

The column detail could be declared as VARCHAR and then input_table could be defined as following:

CREATE TABLE input_table (
    version VARCHAR,
    id VARCHAR,
    detail-type VARCHAR,
    source VARCHAR,
    account VARCHAR,
    time VARCHAR,
    region VARCHAR,
    detail VARCHAR
) with (
    ...
)

Moreover, if you want to process the column detail, you could parse it into a json in a Python UDF as following:

@udf(result_type=DataTypes.STRING())
def get_id(detail):
    detail_json = json.loads(detail)
    if 'build-id' in detail_json:
        return detail_json['build-id']
    else:
        return detail_json['instance-id']

score 3 · Answer 2 · answered Jul 16 '22 at 17:12

I have been searching for a similar functionality (nested-JSON schema definition) in Flink SQL. So far, these are my findings from Flink 1.15.x.

Example JSON

{
   "id":1,
   "name":"Temperature Sensor",
   "payload":{
      "data":{
         "metric":"Temperature",
         "value":23,
      },
      "location":"Berlin",
      "timestamp":"2018-12-10 13:45:00.000"
   }
}

Flink 1.15.x SQL CREATE Statement:

CREATE TABLE sensors (
      id INT,
      name STRING,
      payload STRING
      // Here we declare payload as STRING type since there is no JSON datatype in Apache Flink
    ) WITH (
    'connector' = 'kafka',
    ...
    // Rest of the Apache Kafka connection properties
    ...
);

JSON Functions (Introduced in 1.15.x, released in April '22. The latest is 1.15.1, released July '22)

SELECT * 
FROM sensors 
WHERE JSON_EXISTS(payload, '$.data');

SELECT DISTINCT JSON_VALUE(payload, '$.location') AS `city`
FROM sensors 
WHERE JSON_EXISTS(payload, '$.data');

Unfortunately, I couldn't find a proper way of defining a schema for the nested JSON (in this case, the payload is blindly considered as a string). Would have been better, if that was supported.

Reference:

Does this work? Because on mu case it returns none. – LeparaLaMapara Feb 20 '23 at 11:14 — LeparaLaMapara, Feb 20 '23 at 11:14

How to reference nested JSON within PyFlink SQL when JSON schema varies?

2 Answers2