DataWorks:Offline synchronization of a single table from Kafka to MaxCompute

Step 1: Create a node and configure the task

For the general steps to create and configure a node in the codeless UI, see the Codeless UI configuration guide. This topic does not repeat those steps.

Step 2: Configure the data source and destination

Step 3: Configure field mapping

After you select the data source and destination, you must specify the column mapping between the reader and writer. You can select Map Fields with Same Name, Map Fields in Same Line, Clear Mappings, or Manually Edit Mapping.

• There are six default fields on the Kafka side.

  Field Name	Description
  __key__	The key of the Kafka record.
  __value__	The value of the Kafka record.
  __partition__	The partition number where the Kafka record is located. Partition numbers start from 0.
  __headers__	The headers of the Kafka record.
  __offset__	The offset of the Kafka record in its partition. Offsets start from 0.
  __timestamp__	The 13-digit integer millisecond timestamp of the Kafka record.

• You can configure custom JSON parsing for fields on the Kafka side. Use the . (to retrieve a subfield) and [] (to retrieve an array element) syntax to obtain content from the value field of a JSON-formatted Kafka record.

  Important

  If a JSON field name contains a "." character, you cannot retrieve the field value by defining the field. This is because it causes ambiguity in the field definition syntax.

  The following is an example of the data value of a JSON-formatted record in Kafka.

  {
        "a": {
        "a1": "hello"
        },
        "b": "world",
        "c":[
              "xxxxxxx",
              "yyyyyyy"
              ],
        "d":[
              {
                    "AA":"this",
                    "BB":"is_data"
              },
              {
                    "AA":"that",
                    "BB":"is_also_data"
              }
          ],
       "a.b": "unreachable"
  }

  • To synchronize the data for a1, "hello", you can add the field a.a1 on the Kafka side.

  • To synchronize the data for b, "world", you can add the field b on the Kafka side.

  • To synchronize the data for c, "yyyyyyy", you can add the field c[1] on the Kafka side.

  • To synchronize the data for AA, "this", you can add the field d[0].AA on the Kafka side.

  • If the field on the Kafka side is defined as a.b, you cannot synchronize the data "unreachable".

• Source or destination table fields can be excluded from the mapping. The synchronization instance does not read unmapped source fields. NULL is written to unmapped destination fields.

• You cannot map one source field to multiple destination fields. A destination field also cannot be mapped from multiple source fields.

Step 4: Configure advanced parameters

Click Advanced Configuration on the right side of the task to set parameters such as Maximum Expected Concurrency and Policy for Dirty Data Records. For this tutorial, set the Policy for Dirty Data Records to Ignore Dirty Data Records and leave the other parameters at their default values. For more information, see Codeless UI Configuration.

Step 5: Configure and run a test

1. On the right side of the offline synchronization node editing page, click Debugging Configurations. Set the Resource Group and Script Parameters for the test run. Then, click Run in the top toolbar to test whether the synchronization link runs successfully.

2. You can click the icon in the navigation pane on the left and then click the icon to the right of Personal Directory to create a file with the .sql extension. You can then execute the following SQL query to check whether the data in the destination table is as expected.

   Note

   • To query data in this way, you must attach the destination MaxCompute project to DataWorks as a computing resource.

   • On the .sql file editing page, click Debugging Configurations on the right. Specify the data source Type, Computing Resource, and Resource Group. Then, click Run in the top toolbar.

   SELECT * FROM <MaxCompute_destination_table_name> WHERE pt=<specified_partition> LIMIT 20;

Step 6: Configure scheduling and publish the task

Click Scheduling on the right side of the offline synchronization task. Set the scheduling configuration parameters for the recurring run. Then, click Publish in the top toolbar to open the publishing panel. Follow the on-screen instructions to publish the task.

When you configured the data source and destination, you used three scheduling parameters: ${startTime}, ${endTime}, and ${partition}. In the scheduling configuration, you must specify the replacement policy for these parameters based on your synchronization needs. The following are configuration examples for several typical scenarios.

Set the scheduling cycle based on the desired interval.