OSS-HDFS数据源

使用限制

离线读

• 使用OSS-HDFS Reader时，请注意以下事项：

  由于连通资源组到OSS-HDFS的网络链路比较复杂，建议您使用独享数据集成资源组完成数据同步任务。您需要确保您的独享数据集成资源组具备OSS-HDFS的网络访问能力。

• OSS-HDFS Reader支持以下功能：

  • 支持text、csv、orc和parquet格式的文件，且要求文件内容存放的是一张逻辑意义上的二维表。

  • 支持多种类型数据读取，支持列常量。

  • 支持递归读取、支持通配符*和?。

  • 多个File可以支持并发读取。

离线写

使用OSS-HDFS Writer时，请注意以下事项：

• 目前OSS-HDFS Writer仅支持text、orc和parquet三种格式的文件，且文件内容存放的必须是一张逻辑意义上的二维表。

• 对于text类型，需要保证写入OSS-HDFS文件的分隔符与在Hive上创建表时的分隔符一致，从而实现写入OSS-HDFS数据与Hive表字段关联。

实现过程

支持的字段类型

离线读

ParquetFile、ORCFile、TextFile、CsvFile中的类型，会默认转为数据集成支持的内部类型，如下表所示。

离线写

OSS-HDFS Writer提供向OSS-HDFS文件系统指定路径中写入TextFile文件、 ORCFile文件以及ParquetFile格式文件。

数据同步任务开发

OSS-HDFS数据同步任务的配置入口和通用配置流程指导可参见下文的配置指导，详细的配置参数解释可在配置界面查看对应参数的文案提示。

创建数据源

在进行数据同步任务开发时，您需要在DataWorks上创建OSS-HDFS数据源，操作流程请参见创建与管理数据源。

单表离线同步任务配置指导

• 操作流程请参见通过向导模式配置离线同步任务、通过脚本模式配置离线同步任务。

• 脚本模式配置的全量参数和脚本Demo请参见下文的附录：OSS-HDFS脚本Demo与参数说明。

附录：OSS-HDFS脚本Demo与参数说明

附录：离线任务脚本配置方式

如果您配置离线任务时使用脚本模式的方式进行配置，您需要在任务脚本中按照脚本的统一格式要求编写脚本中的reader参数和writer参数，脚本模式的统一要求请参见通过脚本模式配置离线同步任务，以下为您介绍脚本模式下的数据源的Reader参数和Writer参数的指导详情。

Reader脚本Demo

{
    "type": "job",
    "version": "2.0",
    "steps": [
        {
            "stepType": "oss_hdfs",//插件名
            "parameter": {
                "path": "",//要读取的文件路径
                "datasource": "",//数据源
                "column": [
                    {
                        "index": 0,//序列号,index从0开始（下标索引从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"
                    }
                ],
                "fieldDelimiter": ",",//列分隔符
                "encoding": "UTF-8",//编码格式
                "fileFormat": ""//文本类型
            },
            "name": "Reader",
            "category": "reader"
        },
        {
            "stepType": "stream",
            "parameter": {},
            "name": "Writer",
            "category": "writer"
        }
    ],
    "setting": {
        "errorLimit": {
            "record": ""//错误记录数
        },
        "speed": {
            "concurrent": 3,//作业并发数
            "throttle": true //当throttle值为false时，mbps参数不生效，表示不限流；当throttle值为true时,表示限流。
            "mbps":"12"//限流，此处1mbps = 1MB/s。
        }
    },
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    }
}

Reader脚本参数

{
  "type": "long",
  "index": 0
  //从本地文件文本第一列（下标索引从0开始计数）获取LONG字段，index表示从数据文件中获取列数据。
},
{
  "type": "string",
  "value": "alibaba"
  //HDFS Reader内部生成alibaba的字符串字段作为当前字段，value表示常量列。
}

Writer脚本Demo

{
    "type": "job",
    "version": "2.0",//版本号。
    "steps": [
        {
            "stepType": "stream",
            "parameter": {},
            "name": "Reader",
            "category": "reader"
        },
        {
            "stepType": "oss_hdfs",//插件名。
            "parameter": {
                "path": "",//存储到OSS-HDFS文件系统的路径信息。
                "fileName": "",//OSS-HDFS Writer写入时的文件名。
                "compress": "",//OSS-HDFS文件压缩类型。
                "datasource": "",//数据源。
                "column": [
                    {
                        "name": "col1",//字段名。
                        "type": "string"//字段类型。
                    },
                    {
                        "name": "col2",
                        "type": "int"
                    },
                    {
                        "name": "col3",
                        "type": "double"
                    },
                    {
                        "name": "col4",
                        "type": "boolean"
                    },
                    {
                        "name": "col5",
                        "type": "date"
                    }
                ],
                "writeMode": "",//写入模式。
                "fieldDelimiter": ",",//列分隔符。
                "encoding": "",//编码格式。
                "fileFormat": "text"//文本类型。
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "setting": {
        "errorLimit": {
            "record": ""//错误记录数。
        },
        "speed": {
            "concurrent": 3,//作业并发数。
            "throttle": false //false代表不限流，下面的限流的速度不生效；true代表限流。
        }
    },
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    }
}

Writer脚本参数

{
    "column":
    [
        {
            "name": "userName",
            "type": "string"
        },
        {
            "name": "age",
            "type": "long"
        }
    ]
}

message Message名 {
是否必填 数据类型 列名;
.....................;
}

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;
}