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

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

Okuma yolunu belirtme

path parametresi üç yaklaşımı destekler:

• Seçenek 1: Statik yol — Tek bir dosyayı veya bir dizindeki tüm dosyaları okur. Tek bir dosya bir iş parçacığı kullanır. Örnek: /user/hive/warehouse/mytable01/data.csv.

• Seçenek 2: Joker karakterli yol — Bir desene uyan birden fazla dosyayı okur. HDFS Reader * (herhangi bir karakterle eşleşir) ve ? (tek bir karakterle eşleşir) karakterlerini destekler. Örnek: /hadoop/data_201704*. Gerçek iş parçacığı sayısı, eşleşen dosya sayısı ile concurrent ayarından küçük olanıdır.

• Seçenek 3: Bölüm yolu — Bir Hive bölüm dizininden veri okur. Bir Hive tablosu bölümlerle oluşturulduğunda (örneğin, partition(day="20150820", hour="09")), bölüm HDFS'de bir dizin yapısı olarak görünür. Belirli bir gün için tüm verileri okumak üzere yolu aşağıdaki gibi ayarlayın:

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

Data Integration, bir senkronizasyon görevindeki tüm dosyaları tek bir tablo olarak değerlendirir. Tüm dosyalar aynı şemaya uygun olmalı ve admin hesabında bunlar üzerinde okuma izinleri bulunmalıdır. Dosya adları zamana dayalı bir deseni izliyorsa yolu iş zamanına göre dinamik olarak değiştirmek için zamanlama parametreleri kullanın.

Dosya biçimi ayrıştırma notları

TextFile ve ORCFile karmaşık Hive türlerini farklı şekilde ayrıştırır. map türü için ORCFile {job=80, team=60} üretirken TextFile {job:80, team:60} üretir. Veriler aynıdır ancak biçim farklıdır. Verileriniz karmaşık Hive türleri içeriyorsa yol boyunca tutarlı bir dosya biçimi kullanın. Biçimi birleştirmek için TextFile tablolarını Hive istemcisinde ORCFile'a aktarın.

Parquet şema biçimi

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

• MessageTypeName: İleti türü için bir ad.

• RequiredStatus: Null olmayan sütunlar için required ve null kabul eden sütunlar için optional kullanın. Tüm sütunları optional olarak ayarlayın.

• DataType: Desteklenen türler BOOLEAN, INT32, INT64, INT96, FLOAT, DOUBLE, BINARY (dize türleri için kullanın) ve FIXED_LEN_BYTE_ARRAY şeklindedir.

• Her sütun tanımını sonuncusu dahil olmak üzere noktalı virgülle bitirin.

Örnek:

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

CSV reader yapılandırması

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

Mevcut tüm alanlar ve varsayılanları:

boolean caseSensitive = true;
char textQualifier = 34;
boolean trimWhitespace = true;
boolean useTextQualifier = true;   // Whether to use a CSV escape character
char delimiter = 44;               // Separator
char recordDelimiter = 0;
char comment = 35;
boolean useComments = false;
int escapeMode = 1;
boolean safetySwitch = true;       // Whether to limit a single column to 100,000 characters
boolean skipEmptyRecords = true;   // Whether to skip empty rows
boolean captureRawRecord = true;

Hadoop HA yapılandırması

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

dfs.data.transfer.protection , dfs.datanode.use.datanode.hostname ve dfs.client.use.datanode.hostname parametreleri HDFS Reader eklentisinde Kerberos kimlik doğrulamasını etkinleştirir. HDFS veri kaynağında Kerberos kimlik doğrulaması zaten yapılandırılmışsa bu parametrelerin eklenti yapılandırmasında bulunması gerekmez. HDFS veri kaynağı yapılandırma konusuna bakın.

Kerberos yapılandırma örneği

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

Kerberos, keytab dosyasının mutlak yolunu gerektirdiğinden bu yapılandırmayı bir kaynak grubuna dağıtın.

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

Yazma modları

HDFS Writer yaz-ve-yeniden-adlandır stratejisini kullanır: önce geçici bir klasöre yazar, ardından dosyaları hedef yola taşır. writeMode parametresi, yazma başlamadan önce aynı fileName önekine sahip mevcut dosyaların nasıl ele alınacağını kontrol eder.

Parquet biçimi append modunu desteklemez. Parquet dosyaları için nonConflict kullanın.

OSS destekli HDFS'ye yazma

dataxParquetMode değeri fields olduğunda HDFS Writer, temel depolama olarak OSS'yi destekler. Aşağıdaki OSS parametrelerini hadoopConfig öğesine ekleyin:

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

Aşağıdaki yer tutucuları gerçek değerlerle değiştirin: