Hadoop遷移MaxCompute神器之DataX-On-Hadoop使用指南

MaxCompute Hadoop · 發表 2018-09-11 20:28:07

摘要： DataX-On-Hadoop 即使用hadoop 的任務排程器，將DataX task (Reader->Channel->Writer )排程到hadoop 執行叢集上執行。這樣使用者的hadoop 資料可以通過MR 任...

DataX-On-Hadoop 即使用hadoop 的任務排程器，將DataX task (Reader->Channel->Writer )排程到hadoop 執行叢集上執行。這樣使用者的hadoop 資料可以通過MR 任務批量上傳到MaxCompute 、RDS 等，不需要使用者提前安裝和部署DataX 軟體包，也不需要另外為DataX 準備執行叢集。但是可以享受到DataX 已有的外掛邏輯、流控限速、魯棒重試等等。

1. DataX-On-Hadoop 執行方式

1.1 什麼是DataX-On-Hadoop

DataXofollow,noindex" target="_blank">https://github.com/alibaba/DataX 是阿里巴巴集團內被廣泛使用的離線資料同步工具/平臺，實現包括 SQL/">MySQL、Oracle、HDFS、Hive、OceanBase、HBase、OTS、MaxCompute 等各種異構資料來源之間高效的資料同步功能。 DataX同步引擎內部實現了任務的切分、排程執行能力，DataX的執行不依賴Hadoop環境。

DataX-On-Hadoop是DataX針對Hadoop排程環境實現的版本，使用hadoop的任務排程器，將DataX task(Reader->Channel->Writer)排程到hadoop執行叢集上執行。這樣使用者的hadoop資料可以通過MR任務批量上傳到MaxCompute等，不需要使用者提前安裝和部署DataX軟體包，也不需要另外為DataX準備執行叢集。但是可以享受到DataX已有的外掛邏輯、流控限速、魯棒重試等等。

目前DataX-On-Hadoop支援將Hdfs中的資料上傳到公共雲MaxCompute當中。

1.2 如何執行DataX-On-Hadoop

執行DataX-On-Hadoop步驟如下:

提阿里雲工單申請DataX-On-Hadoop軟體包，此軟體包本質上也是一個Hadoop MR Jar；
通過hadoop客戶端提交一個MR任務，您只需要關係作業的配置檔案內容(這裡是./bvt_case/speed.json，配置檔案和普通的DataX配置檔案完全一致 )，提交命令是：

./bin/hadoop jar datax-jar-with-dependencies.jar com.alibaba.datax.hdfs.odps.mr.HdfsToOdpsMRJob ./bvt_case/speed.json

任務執行完成後，可以看到類似如下日誌:

本例子的Hdfs Reader 和Odps Writer配置資訊如下:

{
"core": {
"transport": {
"channel": {
"speed": {
"byte": "-1",
"record": "-1"
}
}
}
},
"job": {
"setting": {
"speed": {
"byte": 1048576
},
"errorLimit": {
"record": 0
}
},
"content": [
{
"reader": {
"name": "hdfsreader",
"parameter": {
"path": "/tmp/test_datax/big_data*",
"defaultFS": "hdfs://localhost:9000",
"column": [
{
"index": 0,
"type": "string"
},
{
"index": 1,
"type": "string"
}
],
"fileType": "text",
"encoding": "UTF-8",
"fieldDelimiter": ","
}
},
"writer": {
"name": "odpswriter",
"parameter": {
"project": "",
"table": "",
"partition": "pt=1,dt=2",
"column": [
"id",
"name"
],
"accessId": "",
"accessKey": "",
"truncate": true,
"odpsServer": "http://service.odps.aliyun.com/api",
"tunnelServer": "http://dt.odps.aliyun.com",
"accountType": "aliyun"
}
}
}
]
}
}

1.3 DataX-On-Hadoop 任務高階配置引數

針對上面的例子，介紹幾個效能、髒資料的引數:

core.transport.channel.speed.byte 同步任務切分多多個mapper併發執行，每個mapper的同步速度位元Byte 上限，預設為-1，負數表示不限速；如果是1048576表示單個mapper最大速度是1MB/s。
core.transport.channel.speed.record 同步任務切分多多個mapper併發執行，每個mapper的同步速度記錄上限，預設為-1，負數表示不限速；如果是10000表示單個mapper最大記錄速度每秒1萬行。
job.setting.speed.byte 同步任務整體的最大速度，依賴hadoop 2.7.0以後的版本，主要是通過mapreduce.job.running.map.limit引數控制同一時間點mapper的並行度。
job.setting.errorLimit.record 髒資料記錄現在，預設不配置表示不進行髒資料檢查(有髒資料任務不會失敗)；0表示允許髒資料條數最大為0條，如果任務執行時髒資料超過限制，任務會失敗；1表示允許髒資料條數最大為1條，含義不言自明。一個由於髒資料原因失敗的任務:

作業級別的效能引數配置位置示例:

{
"core": {
"transport": {
"channel": {
"speed": {
"byte": "-1",
"record": "-1"
}
}
}
},
"job": {
"setting": {
"speed": {
"byte": 1048576
},
"errorLimit": {
"record": 0
}
},
"content": [
{
"reader": {},
"writer": {}
}
]
}
}

另外，介紹幾個變數替換、作業命名引數:

支援變數引數，比如作業配置檔案json中有如下:

"path": "/tmp/test_datax/dt=${dt}/abc.txt"

任務執行時可以配置如下傳參，使得一份配置程式碼可以多次使用:

./bin/hadoop jar datax-jar-with-dependencies.jar com.alibaba.datax.hdfs.odps.mr.HdfsToOdpsMRJob datax.json -p "-Ddt=20170427 -Dbizdate=123" -t hdfs_2_odps_mr

支援給作業命名，任務執行時的-t引數是作業的traceId，即作業的名字方便根據作業名字即知曉其意圖，比如上面的-t hdfs_2_odps_mr

讀寫外掛詳細配置介紹，請見後續第2、3部分。

2. Hdfs 讀取

2.1 快速介紹

Hdfs Reader提供了讀取分散式檔案系統資料儲存的能力。在底層實現上，Hdfs Reader獲取分散式檔案系統上檔案的資料，並轉換為DataX傳輸協議傳遞給Writer。

Hdfs Reader實現了從Hadoop分散式檔案系統Hdfs中讀取檔案資料並轉為DataX協議的功能。textfile是Hive建表時預設使用的儲存格式，資料不做壓縮，本質上textfile就是以文字的形式將資料存放在hdfs中，對於DataX而言，Hdfs Reader實現上類比TxtFileReader，有諸多相似之處。orcfile，它的全名是Optimized Row Columnar file，是對RCFile做了優化。據官方文件介紹，這種檔案格式可以提供一種高效的方法來儲存Hive資料。Hdfs Reader利用Hive提供的OrcSerde類，讀取解析orcfile檔案的資料。目前Hdfs Reader支援的功能如下：

支援textfile、orcfile、rcfile、sequence file、csv和parquet格式的檔案，且要求檔案內容存放的是一張邏輯意義上的二維表。
支援多種型別資料讀取(使用String表示)，支援列裁剪，支援列常量。
支援遞迴讀取、支援正則表示式（"*"和"?"）。
支援orcfile資料壓縮，目前支援SNAPPY，ZLIB兩種壓縮方式。
支援sequence file資料壓縮，目前支援lzo壓縮方式。
多個File可以支援併發讀取。
csv型別支援壓縮格式有：gzip、bz2、zip、lzo、lzo_deflate、snappy。

我們暫時不能做到：

單個File支援多執行緒併發讀取，這裡涉及到單個File內部切分演算法。後續可以做到支援。

2.2 功能說明

2.2.1 配置樣例

{
"core": {
"transport": {
"channel": {
"speed": {
"byte": "-1048576",
"record": "-1"
}
}
}
},
"job": {
"setting": {
"speed": {
"byte": 1048576
},
"errorLimit": {
"record": 0
}
},
"content": [
{
"reader": {
"name": "hdfsreader",
"parameter": {
"path": "/tmp/test_datax/*",
"defaultFS": "hdfs://localhost:9000",
"column": [
{
"index": 0,
"type": "string"
},
{
"index": 1,
"type": "string"
}
],
"fileType": "text",
"encoding": "UTF-8",
"fieldDelimiter": ","
}
},
"writer": {}
}
]
}
}

2.2.2 引數說明

path
- 描述：要讀取的檔案路徑，如果要讀取多個檔案，可以使用正則表示式"*"，注意這裡可以支援填寫多個路徑。
  
  當指定萬用字元，HdfsReader嘗試遍歷出多個檔案資訊。例如: 指定/*代表讀取/目錄下所有的檔案，指定/yixiao/*代表讀取yixiao目錄下所有的檔案。HdfsReader目前只支援"*"和"?"作為檔案萬用字元。
  
  特別需要注意的是，DataX會將一個作業下同步的所有的檔案視作同一張資料表。使用者必須自己保證所有的File能夠適配同一套schema資訊。並且提供給DataX許可權可讀。
- 必選：是
- 預設值：無
defaultFS
- 描述：Hadoop hdfs檔案系統namenode節點地址。
- 必選：是
- 預設值：無
fileType
- 描述：檔案的型別，目前只支援使用者配置為"text"、"orc"、"rc"、"seq"、"csv"。
  
  text表示textfile檔案格式
  
  orc表示orcfile檔案格式
  
  rc表示rcfile檔案格式
  
  seq表示sequence file檔案格式
  
  csv表示普通hdfs檔案格式（邏輯二維表）
  
  特別需要注意的是，HdfsReader能夠自動識別檔案是orcfile、rcfile、sequence file還是textfile或csv型別的檔案，該項是必填項 ，HdfsReader在做資料同步之前，會檢查使用者配置的路徑下所有需要同步的檔案格式是否和fileType一致，如果不一致則會丟擲異常
  
  另外需要注意的是，由於textfile和orcfile是兩種完全不同的檔案格式，所以HdfsReader對這兩種檔案的解析方式也存在差異，這種差異導致hive支援的複雜複合型別(比如map,array,struct,union)在轉換為DataX支援的String型別時，轉換的結果格式略有差異，比如以map型別為例：
  
  orcfile map型別經hdfsreader解析轉換成datax支援的string型別後，結果為"{job=80, team=60, person=70}"
  
  textfile map型別經hdfsreader解析轉換成datax支援的string型別後，結果為"job:80,team:60,person:70"
  
  從上面的轉換結果可以看出，資料本身沒有變化，但是表示的格式略有差異，所以如果使用者配置的檔案路徑中要同步的欄位在Hive中是複合型別的話，建議配置統一的檔案格式。
  
  如果需要統一複合型別解析出來的格式，我們建議使用者在hive客戶端將textfile格式的表導成orcfile格式的表
column
- 描述：讀取欄位列表，type指定源資料的型別，index指定當前列來自於文字第幾列(以0開始)，value指定當前型別為常量，不從源標頭檔案讀取資料，而是根據value值自動生成對應的列。
  
  預設情況下，使用者可以全部按照string型別讀取資料，配置如下：
```
"column": ["*"]
```
  使用者可以指定column欄位資訊，配置如下：
```
{
"type": "long",
"index": 0//從本地檔案文字第一列獲取int欄位
},
{
"type": "string",
"value": "alibaba"//HdfsReader內部生成alibaba的字串欄位作為當前欄位
}
```
  對於使用者指定column資訊，type必須填寫，index/value必須選擇其一。
- 必選：是
- 預設值：全部按照string型別讀取
fieldDelimiter
- 描述：讀取的欄位分隔符
另外需要注意的是，HdfsReader在讀取textfile資料時，需要指定欄位分割符，如果不指定預設為','，HdfsReader在讀取orcfile時，使用者無需指定欄位分割符，hive本身的預設分隔符為 "\u0001"；若你想將每一行作為目的端的一列，分隔符請使用行內容不存在的字元，比如不可見字元"\u0001" ，分隔符不能使用\n
- 必選：否
- 預設值：,
encoding
- 描述：讀取檔案的編碼配置。
- 必選：否
- 預設值：utf-8
nullFormat
- 描述：文字檔案中無法使用標準字串定義null(空指標)，DataX提供nullFormat定義哪些字串可以表示為null。
  
  例如如果使用者配置: nullFormat:"\N"，那麼如果源頭資料是"\N"，DataX視作null欄位。
- 必選：否
- 預設值：無
compress
- 描述：當fileType（檔案型別）為csv下的檔案壓縮方式，目前僅支援 gzip、bz2、zip、lzo、lzo_deflate、hadoop-snappy、framing-snappy壓縮；值得注意的是，lzo存在兩種壓縮格式：lzo和lzo_deflate，使用者在配置的時候需要留心，不要配錯了；另外，由於snappy目前沒有統一的stream format，datax目前只支援最主流的兩種：hadoop-snappy（hadoop上的snappy stream format）和framing-snappy（google建議的snappy stream format） ;orc檔案型別下無需填寫。
- 必選：否
- 預設值：無

csvReaderConfig

描述：讀取CSV型別檔案引數配置，Map型別。讀取CSV型別檔案使用的CsvReader進行讀取，會有很多配置，不配置則使用預設值。
必選：否

預設值：無

常見配置：

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

所有配置項及預設值,配置時 csvReaderConfig 的map中請嚴格按照以下欄位名字進行配置 ：

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;//單列長度是否限制100000字元
boolean skipEmptyRecords = true;//是否跳過空行
boolean captureRawRecord = true;

hadoopConfig

描述：hadoopConfig裡可以配置與Hadoop相關的一些高階引數，比如HA的配置。

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

必選：否
預設值：無

minInputSplitSize
- 描述：Hadoop hdfs部分檔案型別支援檔案內部切分，這樣一個檔案可以被切分到多個mapper裡面併發執行，每個mapper讀取這個檔案的一部分。這邊測試環境驗證確實可以做到速度的線性擴充套件。注意:由於切分的粒度更細了，啟動mapper數量多可能佔用的機器資源也多一些。目前支援檔案內部切分的檔案型別有: rc、text、csv、parquet
- 必選：否
- 預設值：無限大

2.3 型別轉換

2.3.1 RCFile

如果使用者同步的hdfs檔案是rcfile，由於rcfile底層儲存的時候不同的資料型別儲存方式不一樣，而HdfsReader不支援對Hive元資料資料庫進行訪問查詢，因此需要使用者在column type裡指定該column在hive表中的資料型別，比如該column是bigint型。那麼type就寫為bigint，如果是double型，則填寫double，如果是float型，則填寫float。注意：如果是varchar或者char型別，則需要填寫位元組數，比如varchar(255)，char(30)等，跟hive表中該欄位的型別保持一致，或者也可以填寫string型別。

如果column配置的是*，會讀取所有column，那麼datax會預設以string型別讀取所有column，此時要求column中的型別只能為String,CHAR,VARCHAR中的一種。

RCFile中的型別預設會轉成DataX支援的內部型別，對照表如下：

RCFile在Hive表中的資料型別	DataX 內部型別
TINYINT,SMALLINT,INT,BIGINT	Long
FLOAT,DOUBLE,DECIMAL	Double
String,CHAR,VARCHAR	String
BOOLEAN	Boolean
Date,TIMESTAMP	Date
Binary	Binary

2.3.2 ParquetFile

如果column配置的是*, 會讀取所有列; 此時Datax會預設以String型別讀取所有列. 如果列中出現Double等型別的話, 全部將轉換為String型別。如果column配置讀取特定的列的話, DataX中的型別和Parquet檔案型別的對應關係如下:

Parquet格式檔案的資料型別	DataX 內部型別
int32, int64, int96	Long
float, double	Double
binary	Binary
boolean	Boolean
fixed_len_byte_array	String

textfile,orcfile,sequencefile:

由於textfile和orcfile檔案表的元資料資訊由Hive維護並存放在Hive自己維護的資料庫（如mysql）中，目前HdfsReader不支援對Hive元資料資料庫進行訪問查詢，因此使用者在進行型別轉換的時候，必須指定資料型別，如果使用者配置的column為"*"，則所有column預設轉換為string型別。HdfsReader提供了型別轉換的建議表如下：

DataX 內部型別	Hive表資料型別
Long	TINYINT,SMALLINT,INT,BIGINT
Double	FLOAT,DOUBLE
String	String,CHAR,VARCHAR,STRUCT,MAP,ARRAY,UNION,BINARY
Boolean	BOOLEAN
Date	Date,TIMESTAMP

其中：

Long是指Hdfs檔案文字中使用整形的字串表示形式，例如"123456789"。
Double是指Hdfs檔案文字中使用Double的字串表示形式，例如"3.1415"。
Boolean是指Hdfs檔案文字中使用Boolean的字串表示形式，例如"true"、"false"。不區分大小寫。
Date是指Hdfs檔案文字中使用Date的字串表示形式，例如"2014-12-31"。

特別提醒：

Hive支援的資料型別TIMESTAMP可以精確到納秒級別，所以textfile、orcfile中TIMESTAMP存放的資料類似於"2015-08-21 22:40:47.397898389"，如果轉換的型別配置為DataX的Date，轉換之後會導致納秒部分丟失，所以如果需要保留納秒部分的資料，請配置轉換型別為DataX的String型別。

2.4 按分割槽讀取

Hive在建表的時候，可以指定分割槽partition，例如建立分割槽partition(day="20150820",hour="09")，對應的hdfs檔案系統中，相應的表的目錄下則會多出/20150820和/09兩個目錄，且/20150820是/09的父目錄。瞭解了分割槽都會列成相應的目錄結構，在按照某個分割槽讀取某個表所有資料時，則只需配置好json中path的值即可。

比如需要讀取表名叫mytable01下分割槽day為20150820這一天的所有資料，則配置如下：

"path": "/user/hive/warehouse/mytable01/20150820/*"

3. MaxCompute寫入

3.1 快速介紹

ODPSWriter外掛用於實現往ODPS(即MaxCompute)插入或者更新資料，主要提供給etl開發同學將業務資料匯入MaxCompute，適合於TB,GB數量級的資料傳輸。在底層實現上，根據你配置的專案 / 表 / 分割槽 / 表字段等資訊，通過 Tunnel寫入 MaxCompute 中。支援MaxCompute中以下資料型別：BIGINT、DOUBLE、STRING、DATATIME、BOOLEAN。下面列出ODPSWriter針對MaxCompute型別轉換列表:

DataX 內部型別	MaxCompute 資料型別
Long	bigint
Double	double
String	string
Date	datetime
Boolean	bool

3.2 實現原理

在底層實現上，ODPSWriter是通過MaxCompute Tunnel寫入MaxCompute系統的，有關MaxCompute的更多技術細節請參看 MaxCompute主站:https://www.aliyun.com/product/odps

3.3 功能說明

3.3.1 配置樣例

這裡使用一份從記憶體產生到MaxCompute匯入的資料。

{
"core": {
"transport": {
"channel": {
"speed": {
"byte": "-1048576",
"record": "-1"
}
}
}
},
"job": {
"setting": {
"speed": {
"byte": 1048576
},
"errorLimit": {
"record": 0
}
},
"content": [
{
"reader": {},
"writer": {
"name": "odpswriter",
"parameter": {
"project": "",
"table": "",
"partition": "pt=1,dt=2",
"column": [
"col1",
"col2"
],
"accessId": "",
"accessKey": "",
"truncate": true,
"odpsServer": "http://service.odps.aliyun.com/api",
"tunnelServer": "http://dt.odps.aliyun.com",
"accountType": "aliyun"
}
}
}
]
}
}

3.3.2 引數說明

accessId
- 描述：MaxCompute系統登入ID
- 必選：是
- 預設值：無
accessKey
- 描述：MaxCompute系統登入Key
- 必選：是
- 預設值：無
project
- 描述：MaxCompute表所屬的project，注意:Project只能是字母+數字組合，請填寫英文名稱。在雲端等使用者看到的MaxCompute專案中文名只是顯示名，請務必填寫底層真實地Project英文標識名。
- 必選：是
- 預設值：無
table
- 描述：寫入資料的表名，不能填寫多張表，因為DataX不支援同時匯入多張表。
- 必選：是
- 預設值：無
partition
- 描述：需要寫入資料表的分割槽資訊，必須指定到最後一級分割槽。把資料寫入一個三級分割槽表，必須配置到最後一級分割槽，例如pt=20150101/type＝1/biz=2。
- 必選：如果是分割槽表，該選項必填，如果非分割槽表，該選項不可填寫。
- 預設值：空
column
- 描述：需要匯入的欄位列表，當匯入全部欄位時，可以配置為"column": ["*"], 當需要插入部分MaxCompute列填寫部分列，例如"column": ["id", "name"]。ODPSWriter支援列篩選、列換序，例如表有a,b,c三個欄位，使用者只同步c,b兩個欄位。可以配置成["c","b"], 在匯入過程中，欄位a自動補空，設定為null。
- 必選：否
- 預設值：無
truncate
- 描述：ODPSWriter通過配置"truncate": true，保證寫入的冪等性，即當出現寫入失敗再次執行時，ODPSWriter將清理前述資料，並匯入新資料，這樣可以保證每次重跑之後的資料都保持一致。
  
  truncate選項不是原子操作！MaxCompute SQL無法做到原子性。因此當多個任務同時向一個Table/Partition清理分割槽時候，可能出現併發時序問題，請務必注意！針對這類問題，我們建議儘量不要多個作業DDL同時操作同一份分割槽，或者在多個併發作業啟動前，提前建立分割槽。
- 必選：是
- 預設值：無
odpsServer
- 描述：MaxCompute的server，詳細可以參考文件https://help.aliyun.com/document_detail/34951.html
線上公網地址為http://service.odps.aliyun.com/api
- 必選：是
- 預設值：無
tunnelServer
- 描述：MaxCompute的tunnelserver，詳細可以參考文件https://help.aliyun.com/document_detail/34951.html
線上公網地址為http://dt.odps.aliyun.com
- 必選：是
- 預設值：無
blockSizeInMB
- 描述：為了提高資料寫出MaxCompute的效率ODPSWriter會攢資料buffer，待資料達到一定大小後會進行一次資料提交。blockSizeInMB即為攢資料buffer的大小，預設是64MB。
- 必選：否
- 預設值：64

Hadoop遷移MaxCompute神器之DataX-On-Hadoop使用指南

1. DataX-On-Hadoop 執行方式

1.1 什麼是DataX-On-Hadoop

1.2 如何執行DataX-On-Hadoop

1.3 DataX-On-Hadoop 任務高階配置引數

2. Hdfs 讀取

2.1 快速介紹

2.2 功能說明

2.2.1 配置樣例

2.2.2 引數說明

path

defaultFS

fileType

column

fieldDelimiter

encoding

nullFormat

compress

csvReaderConfig

hadoopConfig

minInputSplitSize

2.3 型別轉換

2.3.1 RCFile

2.3.2 ParquetFile

2.4 按分割槽讀取

3. MaxCompute寫入

3.1 快速介紹

3.2 實現原理

3.3 功能說明

3.3.1 配置樣例

3.3.2 引數說明

accessId

accessKey

project

table

partition

column

truncate

odpsServer

tunnelServer

blockSizeInMB

您可能也會喜歡…