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

Spécification du chemin de lecture

Le paramètre path prend en charge trois approches :

• Option 1 : Chemin statique — Lit un seul fichier ou tous les fichiers d'un répertoire. Un seul fichier utilise un thread. Exemple : /user/hive/warehouse/mytable01/data.csv.

• Option 2 : Chemin avec caractère générique — Lit plusieurs fichiers correspondant à un modèle. HDFS Reader prend en charge * (correspond à n'importe quels caractères) et ? (correspond à un seul caractère). Exemple : /hadoop/data_201704*. Le nombre réel de threads correspond au minimum entre le nombre de fichiers correspondants et la valeur du paramètre concurrent.

• Option 3 : Chemin de partition — Lit les données d'un répertoire de partition Hive. Lorsqu'une table Hive est créée avec des partitions (par exemple, partition(day="20150820", hour="09")), la partition apparaît comme une structure de répertoire dans HDFS. Pour lire toutes les données d'un jour donné, définissez le chemin comme suit :

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

Data Integration traite tous les fichiers d'une tâche de synchronisation comme une seule table. Tous les fichiers doivent respecter le même schéma, et le compte admin doit disposer des autorisations de lecture sur ces fichiers. Si les noms de fichiers suivent un modèle basé sur le temps, utilisez les paramètres de planification pour remplacer dynamiquement le chemin en fonction de l'heure métier.

Notes sur l'analyse des formats de fichier

TextFile et ORCFile analysent différemment les types complexes Hive. Pour le type map, un fichier ORCFile produit {job=80, team=60} tandis qu'un fichier TextFile produit {job:80, team:60}. Les données sont identiques, mais le format diffère. Si vos données incluent des types complexes Hive, utilisez un format de fichier cohérent sur tout le chemin. Pour uniformiser le format, exportez les tables TextFile vers ORCFile dans le client Hive.

Format du schéma Parquet

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

• MessageTypeName : Nom du type de message.

• RequiredStatus : Utilisez required pour les colonnes non nulles et optional pour les colonnes pouvant être nulles. Définissez toutes les colonnes sur optional.

• DataType : Types pris en charge : BOOLEAN, INT32, INT64, INT96, FLOAT, DOUBLE, BINARY (à utiliser pour les types chaîne) et FIXED_LEN_BYTE_ARRAY.

• Terminez chaque définition de colonne par un point-virgule, y compris la dernière.

Exemple :

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

Configuration du lecteur CSV

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

Tous les champs disponibles et leurs valeurs par défaut :

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;

Configuration HA de Hadoop

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

Les paramètres dfs.data.transfer.protection , dfs.datanode.use.datanode.hostname et dfs.client.use.datanode.hostname activent l'authentification Kerberos dans le plug-in HDFS Reader. Si l'authentification Kerberos est déjà configurée sur la source de données HDFS, ces paramètres ne sont pas requis dans la configuration du plug-in. Consultez Configurer une source de données HDFS .

Exemple de configuration Kerberos

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

Comme Kerberos nécessite le chemin absolu vers le fichier keytab, déployez cette configuration sur un groupe de ressources.

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

Modes d'écriture

HDFS Writer utilise une stratégie d'écriture suivie d'un renommage : il écrit d'abord dans un dossier temporaire, puis déplace les fichiers vers le chemin de destination. Le paramètre writeMode contrôle la gestion des fichiers existants avec le même préfixe fileName avant le début de l'écriture.

Le format Parquet ne prend pas en charge le mode append . Utilisez nonConflict pour les fichiers Parquet.

Écriture dans HDFS basé sur OSS

Lorsque dataxParquetMode est fields, HDFS Writer prend en charge OSS comme stockage sous-jacent. Ajoutez les paramètres OSS suivants à hadoopConfig :

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

Remplacez les espaces réservés suivants par des valeurs réelles :