Optimizing Performance of the JSON_VALUE Function in Flink SQL

Scenarios

When the built-in JSON_VALUE function parses multiple fields of a JSON item, the parsing result of the previous JSON item is reused to improve the operator performance.

How to Use

When configuring a Flink job, add custom parameter table.optimizing.json-value-optimization and set it to true on the Flink job development page of the FlinkServer web UI to optimize the performance of the JSON_VALUE function. For details, see Creating a Job.

Example SQL statements:

create table kafkaSource (
    msgValue STRING
) with (
    'connector' = 'kafka'
    ,'topic' = 'topic-1'
    ,'properties.bootstrap.servers' = 'Service IP address of the Kafka Broker instance:Kafka port'
    ,'properties.group.id' = 'group-1'
    ,'value.format' = 'raw'
    ,'scan.startup.mode' = 'group-offsets'
    ,'properties.auto.offset.reset' = 'earliest'
    ,'value.fields-include' = 'EXCEPT_KEY'
);
create table print (
    id STRING,
    name STRING
) with (
      'connector' = 'print'
);
INSERT INTO print
SELECT
    JSON_VAL(msgValue, '$.id'),
    JSON_VAL(msgValue, '$.name')
from kafkaSource;

The IP address and port number of the Kafka Broker instance are as follows:

• To obtain the instance IP address, log in to FusionInsight Manager, choose Cluster > Services > Kafka, click Instances, and query the instance IP address on the instance list page.
• If Kerberos authentication is enabled for the cluster (the cluster is in security mode), the Broker port number is the value of sasl.port. The default value is 21007.
• If Kerberos authentication is disabled for the cluster (the cluster is in normal mode), the Broker port number is the value of port. The default value is 9092. If the port number is set to 9092, set allow.everyone.if.no.acl.found to true. The procedure is as follows:

  Log in to FusionInsight Manager and choose Cluster > Services > Kafka. On the displayed page, click Configurations and then All Configurations. On the displayed page, search for allow.everyone.if.no.acl.found, set it to true, and click Save.