Realtime Compute for Apache Flink:Real-time log ingestion

Prerequisites

• Ensure that your RAM user or RAM role has the required permissions to access the Realtime Compute for Apache Flink console. For more information, see Permissions.

• You have created a Flink workspace. For more information, see Create a workspace.

• Upstream and downstream storage

  • You have created an ApsaraMQ for Kafka instance. For more information, see Step 2: Purchase and deploy an instance.

  • You have created a Hologres instance. For more information, see Purchase a Hologres instance.

  Note

  Your ApsaraMQ for Kafka and Hologres instances must be in the same region and VPC as your Realtime Compute for Apache Flink workspace. Otherwise, you must establish network connectivity. For more information, see Access services across VPCs or Access the internet.

Step 1: Configure an IP whitelist

To allow Flink to access your Kafka and Hologres instances, add the CIDR block of your Flink workspace to the Kafka and Hologres IP whitelists.

1. Obtain the VPC CIDR block of your Flink workspace.

   1. Log in to the Realtime Compute for Apache Flink console.

   2. In the Actions column of the target workspace, choose More > Workspace Details.

   3. In the Workspace Details dialog box, find the VSwitch's CIDR block.

      The dialog box displays basic workspace information and a list of VSwitches. In the CIDR block column, view the CIDR block for each availability zone. Take note of this CIDR block for the next step.

2. Add the CIDR block of the Flink workspace to the IP whitelist of your Kafka instance.

   You must configure the IP whitelist for a VPC endpoint. For detailed steps, see Configure an IP whitelist. In the whitelist editing dialog box, click Add Whitelist IP to add the CIDR block.

3. Add the CIDR block of the Flink workspace to the IP whitelist of your Hologres instance.

   Log in to your Hologres instance and configure its IP whitelist. For detailed steps, see IP whitelist. On the whitelist configuration page in the HoloWeb Security Center, enter the CIDR block in the IP Address field of the Edit IP Whitelist dialog box and click OK.

Step 2: Prepare Kafka test data

Use the Faker connector of Realtime Compute for Apache Flink to generate and write data to ApsaraMQ for Kafka. Follow these steps in the development console of Realtime Compute for Apache Flink.

1. In the ApsaraMQ for Kafka console, create a topic named users.

   For more information, see Create a topic.

2. Create a job to write data to ApsaraMQ for Kafka.

   1. Log on to the development console of Realtime Compute for Apache Flink.

   2. In the Actions column of the target workspace, click Console.

   3. In the left-side navigation pane, choose Development > ETL.

   4. Click the icon, and then click New Stream Draft. Enter a File Name and select an Engine Version.

      Realtime Compute for Apache Flink also offers various code and data synchronization templates. Each includes specific use cases, code samples, and instructions. Click a template to quickly learn the features and syntax of Realtime Compute for Apache Flink and implement your business logic. For more information, see Code templates and Data synchronization templates.

      Parameter	Description	Example
      File Name	The name of the job. Note The job name must be unique within the current workspace.	flink-test
      Engine Version	The Flink engine version for the current job. Select a version labeled Recommended or Stable for higher reliability and performance. For more information about engine versions, see Release notes and Engine versions.	vvr-8.0.8-flink-1.17

   5. Click Create.

   6. Write the SQL statements for the job.

      Copy the following code to the editor and modify the parameters for your environment.

      CREATE TEMPORARY TABLE source (
        id INT,
        first_name STRING,
        last_name STRING,
        `address` ROW<`country` STRING, `state` STRING, `city` STRING>,
        event_time TIMESTAMP
      ) WITH (
        'connector' = 'faker',
        'number-of-rows' = '100',
        'rows-per-second' = '10',
        'fields.id.expression' = '#{number.numberBetween ''0'',''1000''}',
        'fields.first_name.expression' = '#{name.firstName}',
        'fields.last_name.expression' = '#{name.lastName}',
        'fields.address.country.expression' = '#{Address.country}',
        'fields.address.state.expression' = '#{Address.state}',
        'fields.address.city.expression' = '#{Address.city}',
        'fields.event_time.expression' = '#{date.past ''15'',''SECONDS''}'
      );
      
      CREATE TEMPORARY TABLE sink (
        id INT,
        first_name STRING,
        last_name STRING,
        `address` ROW<`country` STRING, `state` STRING, `city` STRING>,
        `timestamp` TIMESTAMP METADATA
      ) WITH (
        'connector' = 'kafka',
        'properties.bootstrap.servers' = 'alikafka-host1.aliyuncs.com:9092,alikafka-host2.aliyuncs.com:9092,alikafka-host3.aliyuncs.com:9092',
        'topic' = 'users',
        'format' = 'json',
        'properties.enable.idempotence'='false'
      );
      
      INSERT INTO sink SELECT * FROM source;

      The following table describes the parameters that you need to modify.

      Parameter	Example value	Description
      properties.bootstrap.servers	alikafka-host1.aliyuncs.com:9092,alikafka-host2.aliyuncs.com:9092,alikafka-host3.aliyuncs.com:9092	The ApsaraMQ for Kafka broker endpoints. A comma-separated list of broker endpoints in 'host:port' format. You can find the VPC Domain Name endpoint in the Endpoint Information section on the Instance Details page.
      topic	users	The ApsaraMQ for Kafka topic name.

3. Start the job.

   1. On the Development > ETL page, click Deploy.

   2. In the Deploy draft dialog box, click Confirm.

   3. Configure resources for the job. For more information, see Configure resources for a job.

   4. On the O&M > Deployments page, find the target deployment and click Start in the Actions column. For more information about the startup configuration, see Start a deployment.

   5. On the Deployments page, you can monitor the runtime information and status of the deployment.

      Because the Faker source generates a bounded stream, the deployment's status changes to FINISHED about one minute after starting. When the deployment is finished, the data has been written to the 'users' topic. The following code shows a sample of the JSON-formatted data.

      {
        "id": 765,
        "first_name": "Barry",
        "last_name": "Pollich",
        "address": {
          "country": "United Arab Emirates",
          "state": "Nevada",
          "city": "Powlowskifurt"
        }
      }

Step 3: Create and start a data sync job

Step 4: View the full synchronization result

1. Log on to the Hologres management console.

2. On the Instances page, click the name of your target instance.

3. In the upper-right corner of the page, click Connect to Instance.

4. On the Metadata Management tab, view the table schema and data of the synchronized users table in the flink_test_db database.

   In the left-side navigation tree, expand your instance name > flink_test_db > test_schema > Tables. The synchronized users table appears.

   The schema and data of the synchronized table are as follows.

   • Table schema

     Double-click the users table name to view the table schema.

     The users table schema contains the following fields: id (BIGINT, primary key), first_name (TEXT), last_name (TEXT), address.country (TEXT), address.state (TEXT), and address.city (TEXT).

     Note

     During full data synchronization, we recommend defining the Kafka metadata partition and offset as the primary key for the Hologres table. This prevents data duplication if the job fails over and retransmits data.

   • Table data

     In the upper-right corner of the users table page, click Query table. Enter the following command and click Run.

     SELECT * FROM test_schema.users;

     The command returns the following result.

     The query returns multiple rows. This confirms that records were successfully synchronized to the users table. Each returned row contains complete data for the id, first_name, last_name, address.country, address.state, and address.city columns.

Step 5: Observe automatic schema synchronization

1. In the ApsaraMQ for Kafka console, manually send a message that contains a new column.

   1. Log in to the ApsaraMQ for Kafka console.

   2. On the Instances page, click the name of the target instance.

   3. On the Topics page, click the name of the target topic (users).

   4. Click Send Message.

   5. Configure the message.

      In the Start to Send and Consume Message dialog box, configure the parameters as follows.

      Parameter	Example
      Method of Sending	Select Console.
      Message Key	Enter flinktest.
      Message Content	Copy and paste the following JSON content into the Message Content field. { "id": 100001, "first_name": "Dennise", "last_name": "Schuppe", "address": { "country": "Isle of Man", "state": "Montana", "city": "East Coleburgh" }, "house-points": { "house": "Pukwudgie", "points": 76 } } Note In this example, house-points is a new nested column.
      Send to Specified Partition	Select Yes.
      Partition ID	Enter 0.

   6. Click OK.

2. In the Hologres console, view the schema and data changes in the users table.

   1. Log in to the Hologres console.

   2. On the Instances page, click the name of the target instance.

   3. In the upper-right corner of the page, click Connect to Instance.

   4. On the Metadata Management tab, double-click the users table name.

   5. Click Query table, enter the following statement, and then click Running.

      SELECT * FROM test_schema.users;

   6. View the query results.

      The query returns the following result.

      The result shows that the record with ID 100001 is successfully written to Hologres, and two new columns, house-points.house and house-points.points, are added to the Hologres table.

      Note

      The message sent to ApsaraMQ for Kafka contains only one nested column: house-points. However, because json.infer-schema.flatten-nested-columns.enable is specified in the WITH clause, Realtime Compute for Apache Flink automatically flattens this column, using the nested field access paths as the new column names.

References

• For information on the CREATE TABLE AS (CTAS) statement, see CREATE TABLE AS (CTAS) statement.

• For information on using Message Queue for Apache Kafka as a source or result table, see Message Queue for Apache Kafka.

• To improve job performance by adjusting node parallelism and resources, see Configure job deployment.