DataWorks:HDFS

HDFS Reader

• Pembacaan konkuren multi-thread pada file tunggal tidak didukung karena algoritma chunking internal. Saat beberapa file ditentukan, HDFS Reader membacanya secara konkuren—jumlah thread aktual adalah nilai terkecil antara jumlah file dan pengaturan concurrent.

• HDFS Reader tidak dapat mengakses metastore Hive. Tentukan tipe data secara eksplisit selama konversi tipe.

• Data TIMESTAMP yang disimpan dalam TextFile dan ORCFile memiliki presisi hingga nanodetik (misalnya, 2015-08-21 22:40:47.397898389). Mengonversinya ke tipe date akan menghilangkan bagian nanodetik. Untuk mempertahankannya, petakan kolom tersebut ke tipe string sebagai gantinya.

• Saat mengonfigurasi tugas sinkronisasi HDFS di editor kode, tugas tersebut tidak memerlukan pengujian konektivitas jaringan yang berhasil untuk sumber data HDFS. Abaikan kesalahan yang muncul.

• Integrasi Data berjalan di bawah akun admin. Akun admin sistem operasi harus memiliki izin baca dan tulis pada file HDFS terkait. Jika tidak, beralihlah ke editor kode dan tambahkan "hdfsUsername": "user_with_permissions" ke skrip.

HDFS Writer

• Hanya format TextFile, ORCFile, dan Parquet yang didukung. Menulis ke RCFile, SequenceFile, atau CSV tidak didukung.

• Menulis ke subset kolom tidak didukung. Karena HDFS adalah sistem file tanpa skema, semua kolom harus ditentukan.

• Tipe data Hive berikut tidak didukung: DECIMAL, BINARY, ARRAY, MAP, STRUCT, dan UNION.

• Untuk tabel partisi Hive, hanya penulisan satu partisi yang didukung.

• Untuk TextFile, delimiter bidang yang digunakan untuk menulis harus sesuai dengan delimiter yang digunakan saat tabel Hive dibuat, agar data dapat dikaitkan dengan bidang tabel Hive.

Kompatibilitas versi plugin

HDFS Reader dan HDFS Writer dibangun di atas Hive 1.1.1 dan Hadoop 2.7.1 (Apache, disesuaikan untuk JDK 1.6 untuk Reader dan JDK 1.7 untuk Writer). Plugin ini telah diuji dengan Hadoop 2.5.0, Hadoop 2.6.0, dan Hive 1.2.0.

Pemetaan tipe HDFS Reader

Secara default, HDFS Reader mengonversi tipe data Hive ke tipe internal Integrasi Data sebagai berikut:

Catatan tentang tipe tertentu:

• `long`: Nilai integer dalam file HDFS, seperti 123456789.

• `double`: Nilai floating-point dalam file HDFS, seperti 3.1415.

• `boolean`: Nilai Boolean (true atau false). Tidak peka huruf besar/kecil.

• `date`: Nilai waktu dalam file HDFS, seperti 2014-12-31 00:00:00.

Pemetaan tipe HDFS Writer

HDFS Writer mendukung tipe data Hive berikut. Konfigurasi kolom harus sesuai dengan tipe kolom yang sesuai di tabel Hive.

Demo skrip Reader

Skrip berikut menunjukkan konfigurasi dasar HDFS Reader. Semua contoh menggunakan parameter datasource untuk mereferensikan sumber data HDFS yang dikonfigurasi di DataWorks.

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

Contoh berikut menunjukkan cara mengonfigurasi HDFS Reader untuk membaca file Parquet menggunakan parquetSchema. Atur fileType ke parquet dan tentukan skema lengkapnya. Gunakan index dalam parameter column untuk memilih dan memetakan kolom yang diperlukan.

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

Parameter 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": false,
    "skipEmptyRecords": false,
    "useTextQualifier": false
}

boolean caseSensitive = true;
char textQualifier = 34;
boolean trimWhitespace = true;
boolean useTextQualifier = true;   // Apakah akan menggunakan karakter escape CSV
char delimiter = 44;               // Pemisah
char recordDelimiter = 0;
char comment = 35;
boolean useComments = false;
int escapeMode = 1;
boolean safetySwitch = true;       // Apakah membatasi satu kolom hingga 100.000 karakter
boolean skipEmptyRecords = true;   // Apakah melewati baris kosong
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@**.**"

Demo skrip Writer

Skrip berikut menunjukkan konfigurasi dasar 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"
            }
        ]
    }
}

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