DDS Sink Stream

Function

DLI outputs the job output data to Document Database Service (DDS).

DDS is compatible with the MongoDB protocol and is secure, highly available, reliable, scalable, and easy to use. It provides DB instance creation, scaling, redundancy, backup, restoration, monitoring, and alarm reporting functions with just a few clicks on the DDS console.

For more information about DDS, see the Document Database Service User Guide.

Prerequisites

• Ensure that you have created a DDS instance on DDS using your account.

  For details about how to create a DDS instance, see Buying a DDS DB Instance in the Document Database Service Getting Started.

• Currently, only cluster instances with SSL authentication disabled are supported. Replica set and single node instances are not supported.
• In this scenario, jobs must run on the dedicated queue of DLI. Ensure that the dedicated queue of DLI has been created.

  To create a dedicated DLI queue, select Pay-per-use for Billing Mode and click Dedicated Resource Mode for Queue Type when creating a queue. For details, see Creating a Queue in the Data Lake Insight User Guide.

• Ensure that a datasource connection has been set up between the DLI dedicated queue and the DDS cluster, and security group rules have been configured based on the site requirements.

  For details about how to create an enhanced datasource connection, see Enhanced Datasource Connections in the Data Lake Insight User Guide.

  For details about how to configure security group rules, see Security Group in the Virtual Private Cloud User Guide.

Syntax

1
2
3
4
5
6
7
8

CREATE SINK STREAM stream_id (attr_name attr_type (',' attr_name attr_type)* )
  WITH (
    type = "dds",
    username = "",
    password = "",
    db_url = "",
    field_names = ""
  );

Keywords

Example

Output data in the qualified_cars stream to the collectionTest DDS DB.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16

CREATE SINK STREAM qualified_cars (
  car_id STRING,
  car_owner STRING,
  car_age INT,
  average_speed INT,
  total_miles INT
)
  WITH (
    type = "dds",
    region = "xxx",
    db_url = "192.168.0.8:8635,192.168.0.130:8635/dbtest/collectionTest",
    username = "xxxxxxxxxx",
    password =  "xxxxxxxxxx",
    field_names = "car_id,car_owner,car_age,average_speed,total_miles",
    batch_insert_data_num = "10"
  );