DataWorks：OSS資料來源

離線讀

OSS Reader實現了從OSS讀取資料並轉為Data Integration協議的功能，OSS本身是無結構化資料存放區。對於Data Integration而言，OSS Reader支援的功能如下。

離線寫

OSS Writer實現了從資料同步協議轉為OSS中的文字檔功能，OSS本身是無結構化資料存放區，目前OSS Writer支援的功能如下。

即時寫

• 支援即時寫入的能力。

• 單表即時支援寫入資料湖：Hudi（0.12.x），Paimon，Iceberg。

單表離線同步任務配置指導

• 操作流程請參見嚮導模式配置、指令碼模式配置。

• 指令碼模式配置的全量參數和指令碼Demo請參見下文的附錄：指令碼Demo與參數說明。

單表即時同步任務配置指導

操作流程請參見Data Integration側即時同步任務配置、DataStudio側即時同步任務配置。

整庫同步配置指導

操作流程請參見整庫離線同步任務、整庫即時同步任務。

離線任務指令碼配置方式

如果您配置離線任務時使用指令碼模式的方式進行配置，您需要按照統一的指令碼格式要求，在任務指令碼中編寫相應的參數，詳情請參見指令碼模式配置，以下為您介紹指令碼模式下資料來源的參數配置詳情。

Reader指令碼Demo：通用樣本

{
    "type":"job",
    "version":"2.0",//版本號碼。
    "steps":[
        {
            "stepType":"oss",//外掛程式名。
            "parameter":{
                "nullFormat":"",//定義可以表示為null的字串。
                "compress":"",//文本壓縮類型。
                "datasource":"",//資料來源。
                "column":[//欄位。
                    {
                        "index":0,//列序號。
                        "type":"string"//資料類型。
                    },
                    {
                        "index":1,
                        "type":"long"
                    },
                    {
                        "index":2,
                        "type":"double"
                    },
                    {
                        "index":3,
                        "type":"boolean"
                    },
                    {
                        "format":"yyyy-MM-dd HH:mm:ss", //時間格式。
                        "index":4,
                        "type":"date"
                    }
                ],
                "skipHeader":"",//類CSV格式檔案可能存在表頭為標題情況，需要跳過。
                "encoding":"",//編碼格式。
                "fieldDelimiter":",",//欄位分隔符號。
                "fileFormat": "",//文本類型。
                "object":[]//object首碼。
            },
            "name":"Reader",
            "category":"reader"
        },
        {
            "stepType":"stream",
            "parameter":{},
            "name":"Writer",
            "category":"writer"
        }
    ],
    "setting":{
        "errorLimit":{
            "record":""//錯誤記錄數。
        },
        "speed":{
            "throttle":true,//當throttle值為false時，mbps參數不生效，表示不限流；當throttle值為true時,表示限流。
            "concurrent":1 //作業並發數。
            "mbps":"12",//限流，此處1mbps = 1MB/s。
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

Reader指令碼Demo：ORC或Parquet檔案讀取OSS

目前通過複用HDFS Reader的方式完成OSS讀取ORC或Parquet格式的檔案，在OSS Reader已有參數的基礎上，增加了Path（ORC）、FileFormat（ORC、Parquet）等擴充配置參數。

• 以ORC檔案格式讀取OSS，樣本如下。

  {
  "stepType": "oss",
  "parameter": {
  "datasource": "",
  "fileFormat": "orc",
  "path": "/tests/case61/orc__691b6815_9260_4037_9899_****",
  "column": [
  {
  "index": 0,
  "type": "long"
  },
  {
  "index": "1",
  "type": "string"
  },
  {
  "index": "2",
  "type": "string"
  }
  ]
  }
  }

• 以Parquet檔案格式讀取OSS，樣本如下。

  {
    "type":"job",
      "version":"2.0",
      "steps":[
      {
        "stepType":"oss",
        "parameter":{
          "nullFormat":"",
          "compress":"",
          "fileFormat":"parquet",
          "path":"/*",
          "parquetSchema":"message m { optional BINARY registration_dttm (UTF8); optional Int64 id; optional BINARY first_name (UTF8); optional BINARY last_name (UTF8); optional BINARY email (UTF8); optional BINARY gender (UTF8); optional BINARY ip_address (UTF8); optional BINARY cc (UTF8); optional BINARY country (UTF8); optional BINARY birthdate (UTF8); optional DOUBLE salary; optional BINARY title (UTF8); optional BINARY comments (UTF8); }",
          "column":[
            {
              "index":"0",
              "type":"string"
            },
            {
              "index":"1",
              "type":"long"
            },
            {
              "index":"2",
              "type":"string"
            },
            {
              "index":"3",
              "type":"string"
            },
            {
              "index":"4",
              "type":"string"
            },
            {
              "index":"5",
              "type":"string"
            },
            {
              "index":"6",
              "type":"string"
            },
            {
              "index":"7",
              "type":"string"
            },
            {
              "index":"8",
              "type":"string"
            },
            {
              "index":"9",
              "type":"string"
            },
            {
              "index":"10",
              "type":"double"
            },
            {
              "index":"11",
              "type":"string"
            },
            {
              "index":"12",
              "type":"string"
            }
          ],
          "skipHeader":"false",
          "encoding":"UTF-8",
          "fieldDelimiter":",",
          "fieldDelimiterOrigin":",",
          "datasource":"wpw_demotest_oss",
          "envType":0,
          "object":[
            "wpw_demo/userdata1.parquet"
          ]
        },
        "name":"Reader",
        "category":"reader"
      },
      {
        "stepType":"odps",
        "parameter":{
          "partition":"dt=${bizdate}",
          "truncate":true,
          "datasource":"0_odps_wpw_demotest",
          "envType":0,
          "column":[
            "id"
          ],
          "emptyAsNull":false,
          "table":"wpw_0827"
        },
        "name":"Writer",
        "category":"writer"
      }
    ],
      "setting":{
      "errorLimit":{
        "record":""
      },
      "locale":"zh_CN",
        "speed":{
        "throttle":false,
          "concurrent":2
      }
    },
    "order":{
      "hops":[
        {
          "from":"Reader",
          "to":"Writer"
        }
      ]
    }
  }

Reader指令碼參數

message MessageType名 {
是否必填, 資料類型, 列名;
......................;
}

"parquetSchema": "message m { optional int32 minute_id; optional int32 dsp_id; optional int32 adx_pid; optional int64 req; optional int64 res; optional int64 suc; optional int64 imp; optional double revenue; }"

"column": ["*"]

"column":
    {
       "type": "long",
       "index": 0    //從OSS文本第一列擷取int欄位。
    },
    {
       "type": "string",
       "value": "alibaba"  //從OSSReader內部產生alibaba的字串欄位作為當前欄位。
    }

Writer指令碼Demo：通用樣本

{
    "type":"job",
    "version":"2.0",
    "steps":[
        {
            "stepType":"stream",
            "parameter":{},
            "name":"Reader",
            "category":"reader"
        },
        {
            "stepType":"oss",//外掛程式名。
            "parameter":{
                "nullFormat":"",//資料同步系統提供nullFormat，定義哪些字串可以表示為null。
                "dateFormat":"",//日期格式。
                "datasource":"",//資料來源。
                "writeMode":"",//寫入模式。
                "writeSingleObject":"false", //表示是否將同步資料寫入單個oss檔案。
                "encoding":"",//編碼格式。
                "fieldDelimiter":","//欄位分隔符號。
                "fileFormat":"",//文本類型。
                "object":""//Object首碼。
            },
            "name":"Writer",
            "category":"writer"
        }
    ],
    "setting":{
        "errorLimit":{
            "record":"0"//錯誤記錄數。
        },
        "speed":{
            "throttle":true,//當throttle值為false時，mbps參數不生效，表示不限流；當throttle值為true時,表示限流。
            "concurrent":1, //作業並發數。
            "mbps":"12"//限流，此處1mbps = 1MB/s。
        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}

Writer指令碼Demo：ORC或Parquet檔案寫入OSS指令碼配置demo

目前通過複用HDFS Writer的方式完成OSS寫ORC或Parquet格式的檔案。在OSS Writer已有參數的基礎上，增加了Path、FileFormat等擴充配置參數，參數含義請參見HDFS Writer。

ORC或Parquet檔案寫入OSS的樣本如下：

• 以ORC檔案格式寫入OSS

  寫ORC檔案，當前僅支援指令碼模式，您需要轉指令碼模式配置，其中fileFormat需要配置為orc，path需要配置為寫入檔案的路徑，column配置格式為 {"name":"your column name","type": "your column type"}。

  當前支援寫入的ORC類型如下：

  欄位類型	離線寫OSS（ORC格式）
  TINYINT	支援
  SMALLINT	支援
  INT	支援
  BIGINT	支援
  FLOAT	支援
  DOUBLE	支援
  TIMESTAMP	支援
  DATE	支援
  VARCHAR	支援
  STRING	支援
  CHAR	支援
  BOOLEAN	支援
  DECIMAL	支援
  BINARY	支援

  {
  "stepType": "oss",
  "parameter": {
  "datasource": "",
  "fileFormat": "orc",
  "path": "/tests/case61",
  "fileName": "orc",
  "writeMode": "append",
  "column": [
  {
  "name": "col1",
  "type": "BIGINT"
  },
  {
  "name": "col2",
  "type": "DOUBLE"
  },
  {
  "name": "col3",
  "type": "STRING"
  }
  ],
  "writeMode": "append",
  "fieldDelimiter": "\t",
  "compress": "NONE",
  "encoding": "UTF-8"
  }
  }

• 以Parquet檔案格式寫入OSS

  {
  "stepType": "oss",
  "parameter": {
  "datasource": "",
  "fileFormat": "parquet",
  "path": "/tests/case61",
  "fileName": "test",
  "writeMode": "append",
  "fieldDelimiter": "\t",
  "compress": "SNAPPY",
  "encoding": "UTF-8",
  "parquetSchema": "message test { required int64 int64_col;\n required binary str_col (UTF8);\nrequired group params (MAP) {\nrepeated group key_value {\nrequired binary key (UTF8);\nrequired binary value (UTF8);\n}\n}\nrequired group params_arr (LIST) {\nrepeated group list {\nrequired binary element (UTF8);\n}\n}\nrequired group params_struct {\nrequired int64 id;\n required binary name (UTF8);\n }\nrequired group params_arr_complex (LIST) {\nrepeated group list {\nrequired group element {\n required int64 id;\n required binary name (UTF8);\n}\n}\n}\nrequired group params_complex (MAP) {\nrepeated group key_value {\nrequired binary key (UTF8);\nrequired group value {\nrequired int64 id;\n required binary name (UTF8);\n}\n}\n}\nrequired group params_struct_complex {\nrequired int64 id;\n required group detail {\nrequired int64 id;\n required binary name (UTF8);\n}\n}\n}",
  "dataxParquetMode": "fields"
  }
  }

Writer指令碼參數

message MessageType名 {
是否必填, 資料類型, 列名;
......................;
}

message m {
optional int64 id;
optional int64 date_id;
optional binary datetimestring;
optional int32 dspId;
optional int32 advertiserId;
optional int32 status;
optional int64 bidding_req_num;
optional int64 imp;
optional int64 click_num;
}