DataWorks：HDFS資料來源

支援的版本

目前不支援阿里雲Apsara File Storage for HDFS版。

使用限制

離線讀

離線寫

使用HDFS Writer時，請注意以下事項：

• 目前HDFS Writer僅支援TextFile、ORCFile和ParquetFile三種格式的檔案，且檔案內容存放的必須是一張邏輯意義上的二維表。

• 由於HDFS是檔案系統，不存在schema的概念，因此不支援對部分列寫入。

• 目前不支援DECIMAL、BINARY、ARRAYS、MAPS、STRUCTS和UNION等Hive資料類型。

• 對於Hive分區表目前僅支援一次寫入單個分區。

• 對於TextFile，需要保證寫入HDFS檔案的分隔字元與在Hive上建立表時的分隔字元一致，從而實現寫入HDFS資料與Hive表欄位關聯。

• 目前外掛程式中的Hive版本為1.1.1，Hadoop版本為2.7.1（Apache為適配JDK1.7）。在Hadoop2.5.0、Hadoop2.6.0和Hive1.2.0測試環境中寫入正常。

• HDFS Writer僅支援使用獨享Data Integration資源群組。

實現過程

支援的欄位類型

離線讀

由於檔案表的中繼資料資訊由Hive維護，並存放在Hive自己維護的中繼資料庫（如MySQL）中。目前HDFS Reader不支援對Hive中繼資料的資料庫進行訪問查詢，因此您在進行類型轉換時，必須指定資料類型。

Hive支援的資料類型TIMESTAMP可以精確到納秒層級，所以TextFile、ORCFile中TIMESTAMP存放的資料類似於2015-08-21 22:40:47.397898389。如果轉換的類型配置為Data Integration的DATE，轉換之後會導致納秒部分丟失。所以如果需要保留納秒部分的資料，請配置轉換類型為Data Integration的字串類型。

離線寫

HDFS Writer提供向HDFS檔案系統指定路徑中寫入TextFile檔案、 ORCFile檔案以及ParquetFile格式檔案，檔案內容可以與Hive中的表關聯。目前HDFS Writer支援大部分Hive類型，請注意檢查您的資料類型。

HDFS Writer針對Hive資料類型的轉換列表，如下所示。

資料同步任務開發

HDFS資料同步任務的配置入口和通用配置流程指導可參見下文的配置指導，詳細的配置參數解釋可在配置介面查看對應參數的文案提示。

建立資料來源

在進行資料同步任務開發時，您需要在DataWorks上建立一個對應的資料來源，操作流程請參見建立並管理資料來源。

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

• 操作流程請參見通過嚮導模式配置離線同步任務、通過指令碼模式配置離線同步任務。

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

附錄：指令碼Demo與參數說明

附錄：離線任務指令碼配置方式

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

Reader指令碼Demo

{
    "type": "job",
    "version": "2.0",
    "steps": [
        {
            "stepType": "hdfs",//外掛程式名
            "parameter": {
                "path": "",//要讀取的檔案路徑
                "datasource": "",//資料來源
                "hadoopConfig":{
                "dfs.data.transfer.protection": "integrity",
               "dfs.datanode.use.datanode.hostname" :"true",
                "dfs.client.use.datanode.hostname":"true"
                 },
                "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",//編碼格式
                "fileType": ""//文本類型
            },
            "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"//限流
        }
    },
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    }
}

"reader":  {
    "name": "hdfsreader",
    "parameter": {
        "path": "/user/hive/warehouse/addata.db/dw_ads_rtb_monitor_minute/thedate=20170103/hour_id=22/*",
        "defaultFS": "h10s010.07100.149:8020",
        "column": [
            {
                "index": 0,
                "type": "string"
            },
            {
                "index": 1,
                "type": "long"
            },
            {
                "index": 2,
                "type": "double"
            }
        ],
        "fileType": "parquet",
        "encoding": "UTF-8",
        "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; }"
    }
}

Reader指令碼參數

{
  "type": "long",
  "index": 0
  //從本地檔案文本第一列（下標索引從0開始計數）擷取INT欄位，index表示從資料檔案中擷取列資料。
},
{
  "type": "string",
  "value": "alibaba"
  //HDFS Reader內部產生alibaba的字串欄位作為當前欄位，value表示常量列。
}

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

"csvReaderConfig":{
  "safetySwitch": false,
  "skipEmptyRecords": false,
  "useTextQualifier": false
}

boolean caseSensitive = true;
char textQualifier = 34;
boolean trimWhitespace = true;
boolean useTextQualifier = true;//是否使用csv逸出字元。
char delimiter = 44;//分隔字元
char recordDelimiter = 0;
char comment = 35;
boolean useComments = false;
int escapeMode = 1;
boolean safetySwitch = true;//單列長度是否限制100,000字元。
boolean skipEmptyRecords = true;//是否跳過空行。
boolean captureRawRecord = true;

"hadoopConfig":{
"dfs.nameservices": "testDfs",
"dfs.ha.namenodes.testDfs": "namenode1,namenode2",
"dfs.namenode.rpc-address.youkuDfs.namenode1": "",
"dfs.namenode.rpc-address.youkuDfs.namenode2": "",
"dfs.client.failover.proxy.provider.testDfs": "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider",
"dfs.data.transfer.protection": "integrity",
"dfs.datanode.use.datanode.hostname" :"true",
"dfs.client.use.datanode.hostname":"true"
}

"hadoopConfig":{ "dfs.data.transfer.protection": "integrity", "dfs.datanode.use.datanode.hostname" :"true", "dfs.client.use.datanode.hostname":"true" }

"haveKerberos":true,
"kerberosKeytabFilePath":"/opt/datax/**.keytab",
"kerberosPrincipal":"**/hadoopclient@**.**"

Writer指令碼Demo

{
    "type": "job",
    "version": "2.0",//版本號碼。
    "steps": [
        { 
            "stepType": "stream",
            "parameter": {},
            "name": "Reader",
            "category": "reader"
        },
        {
            "stepType": "hdfs",//外掛程式名。
            "parameter": {
                "path": "",//儲存到Hadoop HDFS檔案系統的路徑資訊。
                "fileName": "",//HDFS Writer寫入時的檔案名稱。
                "compress": "",//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": "",//編碼格式。
                "fileType": "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 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;
}

"hadoopConfig":{
"dfs.nameservices": "testDfs",
"dfs.ha.namenodes.testDfs": "namenode1,namenode2",
"dfs.namenode.rpc-address.youkuDfs.namenode1": "",
"dfs.namenode.rpc-address.youkuDfs.namenode2": "",
"dfs.client.failover.proxy.provider.testDfs": "org.apache.hadoop.hdfs.server.namenode.ha.ConfiguredFailoverProxyProvider"
}

```json
    "writer": {
    "name": "hdfswriter",
    "parameter": {
        "defaultFS": "oss://test-bucket",
        "fileType": "parquet",
        "path": "/datasets/oss_demo/kpt",
        "fileName": "test",
        "writeMode": "truncate",
        "encoding": "UTF-8",
        "hadoopConfig": {
            "fs.oss.accessKeyId": "the-access-id",
            "fs.oss.accessKeySecret": "the-access-key",
            "fs.oss.endpoint": "oss-cn-hangzhou.aliyuncs.com"
            },
            "parquetSchema": "message test {\n    required int64 id;\n    optional binary name (UTF8);\n    optional int64 gmt_create;\n    required group map_col (MAP) {\n        repeated group key_value {\n            required binary key (UTF8);\n            required binary value (UTF8);\n        }\n    }\n    required group array_col (LIST) {\n        repeated group list {\n            required binary element (UTF8);\n        }\n    }\n    required group struct_col {\n        required int64 id;\n        required binary name (UTF8);\n    }    \n}",
            "dataxParquetMode": "fields"
            }
        }
    ```

"haveKerberos":true,
"kerberosKeytabFilePath":"/opt/datax/**.keytab",
"kerberosPrincipal":"**/hadoopclient@**.**"