Set up the basic environment

Background

Step 1: Create the basic environment with CADT

1. Log on to the CADT console and create an application.

   1. Log on to the CADT console.

   2. From the top menu bar, choose New > Official Solution > Template.

   3. In the search box, enter "Flink migration" and press Enter.

   4. Hover over Migrate open source Flink to Realtime Compute for Apache Flink and click Create from Solution.

2. (Optional) Change the deployment region.

   1. Double-click the outer border of the canvas.

   2. In the right-side panel, select a region.

      On the Single-region and single-zone details panel, select a region from the Region drop-down list, such as China (Shanghai).

   3. After changing the region, click Confirm Modification. A confirmation dialog appears, prompting you to reconfigure the vSwitch zone for the new region.

3. Change the Flink cluster password.

   1. Double-click the My-FLINK cluster icon.

   2. In the right-side panel, click Clear and then enter a new password.

      In the E-MapReduce on ECS details panel, set Logon Method to Set Password. Click Clear next to Logon Password and Confirm Logon Password, then enter a new password.

4. Modify the Message Queue for Apache Kafka instance configuration.

   1. Double-click the kafka instance icon.

   2. Change Deployment Instance-Version to the latest version, 2.6.2.

      In the Message Queue for Apache Kafka details panel, find the Deployment Instance-Version drop-down list and select 2.6.2.

   3. Configure the topic.

      In this example, change the topic name to kafka-order.

      Set Topic Description to kafka-order and Partitions to 12. Keep the default values for Storage Engine and Message Type.

5. Modify the ApsaraDB RDS configuration.

   1. Double-click the flink-rds instance icon.

   2. Configure the database account.

      Enter the database account, account type, and password.

      For example, set Database Account to flink_rds, set Account Type to Standard Account, and then set a password.

   3. Configure the database.

      Enter a database name, and select an authorized account and a database authorization mode.

      For example, set Database Name to test_db, set Supported Character Sets to utf8, set Authorized Account to flink_rds, and set Database Authorization to Read/Write.

6. Deploy the application.

   1. In the upper-right corner of the page, click Save. Enter an application name and click OK.

   2. Click Switch to Edit Mode.

      After saving the template, the page switches to browse mode. You must switch back to edit mode to deploy the application.

   3. Click Deploy Application.

      The top of the page shows three deployment stages: Resource Validation, Resource Pricing, and Resource Deployment.

   4. Click Next: Price List.

   5. Click Next: Confirm Order.

   6. Select Terms of Service for Cloud Architecture Design Tool and click Next: Pay and Create.

      The Confirm Order page lists the resources to be created, including My-FLINK (E-MapReduce on ECS), kafka (Message Queue for Apache Kafka), flink-rds (ApsaraDB RDS), and oss-flink-cadt (Object Storage Service (OSS)). Confirm the billing information and complete the payment.

      After deployment, the Deployment Successful indicator appears in the upper-left corner of the page. In the CADT architecture diagram, a green check mark on each component (kafka, My-FLINK, flink-rds, flink-on-cloud, oss-flink-namech, and so on) indicates that all cloud resources have been created and deployed.

Step 2: Test data generation with a Flink SQL job

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

   1. Double-click the flink-on-cloud icon.

      The flink-on-cloud component sits at the bottom of the CADT architecture diagram and represents the Realtime Compute for Apache Flink fully managed instance.

   2. Click Go to Console.

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

      On the Fully Managed Flink tab of the Realtime Compute console, locate the workspace named flink-on-cloud. Its Workspace Status is Running, Processor Architecture is X86, and Billing Method is Pay-As-You-Go.

2. Create a Flink SQL streaming job.

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

   2. Click the icon, and then click New Streaming Job. Enter a File Name, select an Engine Version, and click Create.

   3. On the SQL job development page, write the DDL and DML code.

      Sample code:

      CREATE TEMPORARY TABLE data_in (
        id VARCHAR, 
        order_value FLOAT
      ) WITH (
        'connector' = 'datagen',
        'rows-per-second' = '100',
        'fields.id.length' = '10',
        'fields.order_value.min' = '1.0',
        'fields.order_value.max' = '100.0'
      );
      
      CREATE TEMPORARY TABLE kafka_order (
        order_id VARCHAR,
        order_time TIMESTAMP,
        order_type VARCHAR,
        order_value FLOAT
      ) WITH (
        'connector' = 'kafka',
        'topic' = 'kafka-order',
        'properties.bootstrap.servers' = 'alikafka-****.aliyuncs.com:9092,alikafka-****.aliyuncs.com:9092',
        'format' = 'csv'
      );
      
      INSERT INTO kafka_order
      SELECT id as order_id,
        CURRENT_TIMESTAMP as order_time,
        CASE
          WHEN (substring (id, 1, 1) <= 'z' AND substring (id, 1, 1) >= 'a') and (substring (id, 2, 1) <= 'z' AND substring (id, 2, 1) >= 'a') THEN 'typeA'
          WHEN (substring (id, 1, 1) <= 'z' AND substring (id, 1, 1) >= 'a') and (substring (id, 2, 1) <= '9' AND substring (id, 2, 1) >= '0') THEN 'typeB'
          WHEN (substring (id, 1, 1) <= '9' AND substring (id, 1, 1) >= '0') and (substring (id, 2, 1) <= '9' AND substring (id, 2, 1) >= '0') THEN 'typeC'
          ELSE 'typeD'
        END as order_type,
        order_value
      FROM
        data_in;

3. In the SQL code, replace the value of properties.bootstrap.servers with your Kafka instance's endpoint.

   1. On the CADT template page, double-click the kafka instance icon.

   2. Click Go to Console to view the Endpoint Information.

      On the Kafka instance's Instance Details page, locate the Endpoint Information section. For the default endpoint, Network is VPC and Protocol is PLAINTEXT. Copy the Domain Endpoint address to configure the properties.bootstrap.servers parameter in the SQL code.

4. In the upper-right corner of the SQL editor, click Deploy. In the Deploy New Version dialog box, click OK.

5. On the O&M > Job O&M page, find the target job and click Start in the Actions column.

6. In the Start Job dialog box, keep the default option Stateless Start and click Start.

Step 3: View test data in the Kafka console

1. Log on to the Message Queue for Apache Kafka console.

   1. On the CADT template page, double-click the kafka instance icon.

   2. Click Go to Console.

      After you select the kafka instance, its resource information appears at the bottom of the CADT page.

2. In the Kafka topic, view the messages written by Realtime Compute for Apache Flink.

   1. In the left-side navigation pane, click Topic management.

   2. Click kafka_order.

   3. On the Message query tab, query for messages.

      Select Query by Time, set Partition to All Partitions, specify a Time Point, and click Query. The result table shows fields including Partition, Offset, Key, Value, and Message Creation Time. The Value column contains the CSV-formatted test messages written by Flink.

      Note

      After the Flink job starts running, the test data appears in the Kafka topic.