Function

Create a source stream to read data from DIS. DIS accesses user data and Flink job reads data from the DIS stream as input data for jobs. Flink jobs can quickly remove data from producers using DIS source sources for continuous processing. Flink jobs are applicable to scenarios where data outside the cloud service is imported to the cloud service for filtering, real-time analysis, monitoring reports, and dumping.

DIS addresses the challenge of transmitting data outside cloud services to cloud services. DIS builds data intake streams for custom applications capable of processing or analyzing streaming data. DIS continuously captures, transmits, and stores terabytes of data from hundreds of thousands of sources every hour, such as logs, Internet of Things (IoT) data, social media feeds, website clickstreams, and location-tracking events. For more information about DIS, see the Data Ingestion Service User Guide.

Syntax

CREATE SOURCE STREAM stream_id (attr_name attr_type (',' attr_name attr_type)* )
  WITH (
    type = "dis",
    region = "",
    channel = "",
    partition_count = "",
    encode = "",
    field_delimiter = "",
    offset= "")
  (TIMESTAMP BY timeindicator (',' timeindicator)?);timeindicator:PROCTIME '.' PROCTIME| ID '.' ROWTIME

Keyword

Precautions

The attribute type used as the timestamp must be long or timestamp.

Example

• In CSV encoding format, DLI reads data from the DIS stream and records it as codes in CSV format. The codes are separated by commas (,).

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

  CREATE SOURCE STREAM car_infos ( car_id STRING, car_owner STRING, car_age INT, average_speed INT, total_miles INT, car_timestamp LONG ) WITH ( type = "dis", region = "cn-north-1" , channel = "dliinput", encode = "csv", field_delimiter = "," ) TIMESTAMP BY car_timestamp.rowtime;

• In JSON encoding format, DLI reads data from the DIS stream and records it as codes in JSON format. For example, {"car":{"car_id":"ZJA710XC", "car_owner":"coco", "car_age":5, "average_speed":80, "total_miles":15000, "car_timestamp":1526438880}}

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

  CREATE SOURCE STREAM car_infos ( car_id STRING, car_owner STRING, car_age INT, average_speed INT, total_miles INT, car_timestamp LONG ) WITH ( type = "dis", region = "cn-north-1" , channel = "dliinput", encode = "json", json_config = "car_id=car.car_id;car_owner =car.car_owner;car_age=car.car_age;average_speed =car.average_speed ;total_miles=car.total_miles;" ) TIMESTAMP BY car_timestamp.rowtime;

• In XML encoding format, DLI reads data from the DIS stream and records it as codes in XML format.

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

  CREATE SOURCE STREAM person_infos ( pid BIGINT, pname STRING, page int, plocation STRING, pbir DATE, phealthy BOOLEAN, pgrade ARRAY[STRING] ) WITH ( type = "dis", region = "cn-north-1" , channel = "dis-dli-input", encode = "xml", field_delimiter = ",", xml_config = "pid=person.pid;page=person.page;pname=person.pname;plocation=person.plocation;pbir=person.pbir;pgrade=person.pgrade;phealthy=person.phealthy" );

  An example of XML data is as follows:

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

  <?xml version="1.0" encodeing="utf-8"?> <root> <person> <pid>362305199010025042</pid> <pname>xiaoming</pname> <page>28</page> <plocation>Shangdu County, Jining District, Ulanchap, Inner Mongolia</plocation> <pbir>1990-10-02</pbir> <phealthy>true</phealthy> <pgrade>[A,B,C]</pgrade> </person> </root>

• In EMAIL encoding format, DLI reads data from the DIS stream and records it as a complete Email.

  1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21

  CREATE SOURCE STREAM email_infos ( Event_ID String, Event_Time Date, Subject String, From_Email String, To_EMAIL String, CC_EMAIL Array[String], BCC_EMAIL String, MessageBody String, Mime_Version String, Content_Type String, charset String, Content_Transfer_Encoding String ) WITH ( type = "dis", region = "cn-north-1" , channel = "dliinput", encode = "email", email_key = "Message-ID, Date, Subject, From, To, CC, BCC, Body, Mime-Version, Content-Type, charset, Content_Transfer_Encoding" );

  An example of email data is as follows:

  Message-ID: <200906291839032504254@sample.com>
  Date: Fri, 11 May 2001 09:54:00 -0700 (PDT)
  From: zhangsan@sample.com
  To: lisi@sample.com, wangwu@sample.com 
  Subject:  "Hello World" 
  Cc: lilei@sample.com, hanmei@sample.com
  Mime-Version: 1.0
  Content-Type: text/plain; charset=us-ascii
  Content-Transfer-Encoding: 7bit
  Bcc: jack@sample.com, lily@sample.com
  X-From: Zhang San
  X-To: Li Si, Wang Wu
  X-cc: Li Lei, Han Mei
  X-bcc: 
  X-Folder: \Li_Si_June2001\Notes Folders\Notes inbox
  X-Origin: Lucy
  X-FileName: sample.nsf
  
  Dear Associate / Analyst Committee:
  
  Hello World! 
  
  Thank you,
  
  Associate / Analyst Program 
  zhangsan