code: def process_json_data():
      env = StreamExecutionEnvironment.get_execution_environment()
  
      # define the source
      ds = env.from_collection(
          collection=[
              (
                  1,
                  '{"name": "Flink", "tel": 123, "addr": {"country": "Germany", "city": "Berlin"}}',
              ),
              (
                  2,
                  '{"name": "hello", "tel": 135, "addr": {"country": "China", "city": "Shanghai"}}',
              ),
              (
                  3,
                  '{"name": "world", "tel": 124, "addr": {"country": "USA", "city": "NewYork"}}',
              ),
              (
                  4,
                  '{"name": "PyFlink", "tel": 32, "addr": {"country": "China", "city": "Hangzhou"}}',
              ),
          ]
      )
  
      def update_tel(data):
          # parse the json
          json_data = json.loads(data[1])
          json_data["tel"] += 1
          return data[0], json_data
  
      def filter_by_country(data):
          # the json data could be accessed directly, there is no need to parse it again using
          # json.loads
          return "China" in data[1]["addr"]["country"]
  
      ds.map(update_tel).filter(filter_by_country).print()
  
      # submit for execution
      env.execute()
  
imports: import json
  import logging
  import sys
  from pyflink.datastream import StreamExecutionEnvironment
label_desc: Use image node_id 1 for calling a global function with example usage: process_json_data() and returns: data,  json_data, unknown