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

Especificación de la ruta de lectura

El parámetro path admite tres enfoques:

• Opción 1: Ruta estática — Lee un solo archivo o todos los archivos de un directorio. Un solo archivo utiliza un hilo. Ejemplo: /user/hive/warehouse/mytable01/data.csv.

• Opción 2: Ruta con comodines — Lee varios archivos que coinciden con un patrón. HDFS Reader admite * (coincide con cualquier carácter) y ? (coincide con un solo carácter). Ejemplo: /hadoop/data_201704*. La cantidad real de hilos corresponde al valor menor entre el número de archivos coincidentes y la configuración concurrent.

• Opción 3: Ruta de partición — Lee datos de un directorio de partición de Hive. Cuando se crea una tabla de Hive con particiones (por ejemplo, partition(day="20150820", hour="09")), la partición aparece como una estructura de directorios en HDFS. Para leer todos los datos de un día determinado, establezca la ruta de la siguiente manera:

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

Data Integration trata todos los archivos de una tarea de sincronización como una sola tabla. Todos los archivos deben cumplir con el mismo esquema y la cuenta admin debe tener permisos de lectura sobre ellos. Si los nombres de archivo siguen un patrón basado en tiempo, utilice scheduling parameters para reemplazar la ruta dinámicamente según el tiempo del negocio.

Notas sobre el análisis de formatos de archivo

TextFile y ORCFile analizan los tipos complejos de Hive de manera diferente. Para el tipo map, un ORCFile produce {job=80, team=60}, mientras que un TextFile produce {job:80, team:60}. Los datos son los mismos, pero el formato difiere. Si sus datos incluyen tipos complejos de Hive, utilice un formato de archivo consistente en toda la ruta. Para unificar el formato, exporte las tablas TextFile a ORCFile en el cliente de Hive.

Formato de esquema Parquet

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

• MessageTypeName: Un nombre para el tipo de mensaje.

• RequiredStatus: Utilice required para columnas no nulas y optional para columnas que aceptan valores nulos. Establezca todas las columnas como optional.

• DataType: Los tipos admitidos son BOOLEAN, INT32, INT64, INT96, FLOAT, DOUBLE, BINARY (utilizar para tipos de cadena) y FIXED_LEN_BYTE_ARRAY.

• Finalice cada definición de columna con punto y coma, incluida la última.

Ejemplo:

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

Configuración del lector CSV

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

Todos los campos disponibles y sus valores predeterminados:

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;

Configuración de 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"
}

Los parámetros dfs.data.transfer.protection , dfs.datanode.use.datanode.hostname y dfs.client.use.datanode.hostname habilitan la autenticación Kerberos en el plugin HDFS Reader. Si la autenticación Kerberos ya está configurada en el origen de datos HDFS, estos parámetros no son necesarios en la configuración del plugin. Consulte Configure an HDFS data source .

Ejemplo de configuración de Kerberos

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

Dado que Kerberos requiere la ruta absoluta al archivo keytab, despliegue esta configuración en un resource group.

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

Modos de escritura

HDFS Writer utiliza una estrategia de escritura y renombrado: primero escribe en una carpeta temporal y luego mueve los archivos a la ruta de destino. El parámetro writeMode controla cómo se manejan los archivos existentes con el mismo prefijo fileName antes de comenzar la escritura.

El formato Parquet no admite el modo append . Utilice nonConflict para archivos Parquet.

Escritura en HDFS respaldado por OSS

Cuando dataxParquetMode es fields, HDFS Writer admite OSS como almacenamiento subyacente. Agregue los siguientes parámetros de OSS a 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"
    }
}

Reemplace los siguientes marcadores de posición con los valores reales: