大数据开发治理平台 DataWorks：OSS数据源

{
    "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"
            }
        ]
    }
}

目前通过复用HDFS Reader的方式完成OSS读取ORC或Parquet格式的文件，在OSS Reader已有参数的基础上，增加了Path、FileFormat等扩展配置参数。

• 以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"
        }
      ]
    }
  }

{
    "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"
            }
        ]
    }
}

目前通过复用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"
  }
  }