DataWorks:HDFS

HDFS リーダー

• 内部のチャンク分割アルゴリズムのため、単一ファイル のマルチスレッドによる並列読み取りはサポートされていません。複数のファイルを指定した場合、HDFS リーダーはそれらを並列に読み取ります。実際のスレッド数は、ファイル数と concurrent 設定のいずれか小さい方の値になります。

• HDFS リーダーは Hive のメタストアにアクセスできません。型変換時にデータ型を明示的に指定してください。

• TextFile および ORCFile に格納された TIMESTAMP データはナノ秒精度です (例： 2015-08-21 22:40:47.397898389) 。 date 型に変換すると、ナノ秒部分が切り捨てられます。ナノ秒部分を保持するには、代わりに列を string 型にマッピングしてください。

• コードエディタで HDFS の同期タスクを設定する場合、HDFS データソースへのネットワーク接続テストが成功する必要はありません。発生したエラーは無視してください。

• Data Integration は admin アカウントで実行されます。オペレーティングシステムの admin アカウントには、該当する HDFS ファイルへの読み取りおよび書き込み権限が必要です。権限がない場合は、コードエディタに切り替え、スクリプトに "hdfsUsername": "user_with_permissions" を追加してください。

HDFS ライター

• サポートされているのは、TextFile、ORCFile、および Parquet 形式のみです。RCFile、SequenceFile、または CSV への書き込みはサポートされていません。

• 一部の列への書き込みはサポートされていません。HDFS はスキーマを持たないファイルシステムであるため、すべての列を指定する必要があります。

• 次の Hive データ型はサポートされていません： DECIMAL、BINARY、ARRAY、MAP、STRUCT、および UNION。

• Hive のパーティションテーブルでは、単一パーティションへの書き込みのみがサポートされています。

• TextFile の場合、書き込みに使用するフィールドデリミタは、Hive テーブルの作成時に使用したデリミタと一致させる必要があります。これにより、データを Hive テーブルのフィールドに関連付けることができます。

プラグインバージョンの互換性

HDFS リーダーと HDFS ライターは、Hive 1.1.1 および Hadoop 2.7.1 (Apache、リーダーは JDK 1.6 に、ライターは JDK 1.7 に対応) をベースに構築されています。これらのプラグインは、Hadoop 2.5.0、Hadoop 2.6.0、および Hive 1.2.0 でテスト済みです。

HDFS リーダーのタイプマッピング

デフォルトでは、HDFS リーダーは Hive データ型を、Data Integration の内部データ型に次のように変換します。

特定のデータ型に関する注記：

• long：HDFS ファイル内の整数値 (例：123456789)。

• double：HDFS ファイル内の浮動小数点数 (例：3.1415)。

• boolean：ブール値 (true または false)。大文字と小文字を区別しません。

• date：HDFS ファイル内の時刻値 (例：2014-12-31 00:00:00)。

HDFS ライターのタイプマッピング

HDFS ライターは、以下の Hive データ型をサポートしています。カラム設定は、Hive テーブルの対応するカラムのデータ型と一致させる必要があります。

Reader スクリプトのデモ

次のスクリプトは、基本的な HDFS Reader の設定を示しています。すべての例では、datasource パラメーターを使用して、DataWorks で設定された HDFS データソースを参照します。

{
    "type": "job",
    "version": "2.0",
    "steps": [
        {
            "stepType": "hdfs",
            "parameter": {
                "datasource": "",
                "path": "",
                "fileType": "",
                "column": [
                    {
                        "index": 0,
                        "type": "string"
                    },
                    {
                        "index": 1,
                        "type": "long"
                    },
                    {
                        "index": 2,
                        "type": "double"
                    },
                    {
                        "index": 3,
                        "type": "boolean"
                    },
                    {
                        "index": 4,
                        "type": "date",
                        "format": "yyyy-MM-dd HH:mm:ss"
                    }
                ],
                "fieldDelimiter": ",",
                "encoding": "UTF-8",
                "hadoopConfig": {
                    "dfs.data.transfer.protection": "integrity",
                    "dfs.datanode.use.datanode.hostname": "true",
                    "dfs.client.use.datanode.hostname": "true"
                }
            },
            "name": "Reader",
            "category": "reader"
        },
        {
            "stepType": "stream",
            "parameter": {},
            "name": "Writer",
            "category": "writer"
        }
    ],
    "setting": {
        "errorLimit": {
            "record": ""
        },
        "speed": {
            "concurrent": 3,
            "throttle": true,
            "mbps": "12"
        }
    },
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    }
}

次の例は、parquetSchema を使用して HDFS Reader を設定し、Parquet ファイルを読み取る方法を示しています。fileType を parquet に設定し、完全なスキーマを指定します。column パラメーター内の index を使用して、必要な列を選択し、マッピングします。

"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",
        "fileType": "parquet",
        "encoding": "UTF-8",
        "column": [
            {
                "index": 0,
                "type": "string"
            },
            {
                "index": 1,
                "type": "long"
            },
            {
                "index": 2,
                "type": "double"
            }
        ],
        "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 のパラメーター

message MessageTypeName {
    RequiredStatus DataType ColumnName;
    ...;
}

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

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

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.testDfs.namenode1": "",
    "dfs.namenode.rpc-address.testDfs.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"
}

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

Writer スクリプトのデモ

次のスクリプトは、基本的な HDFS Writer の設定を示しています。

{
    "type": "job",
    "version": "2.0",
    "steps": [
        {
            "stepType": "stream",
            "parameter": {},
            "name": "Reader",
            "category": "reader"
        },
        {
            "stepType": "hdfs",
            "parameter": {
                "datasource": "",
                "path": "",
                "fileName": "",
                "fileType": "text",
                "column": [
                    {
                        "name": "col1",
                        "type": "string"
                    },
                    {
                        "name": "col2",
                        "type": "int"
                    },
                    {
                        "name": "col3",
                        "type": "double"
                    },
                    {
                        "name": "col4",
                        "type": "boolean"
                    },
                    {
                        "name": "col5",
                        "type": "date"
                    }
                ],
                "writeMode": "",
                "fieldDelimiter": ",",
                "encoding": "UTF-8",
                "compress": ""
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "setting": {
        "errorLimit": {
            "record": ""
        },
        "speed": {
            "concurrent": 3,
            "throttle": false
        }
    },
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    }
}

Writer のパラメーター

"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": "<your-access-key-id>",
            "fs.oss.accessKeySecret": "<your-access-key-secret>",
            "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"
    }
}