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

Especificando o caminho de leitura

O parâmetro path aceita três abordagens:

• Opção 1: Caminho estático — Lê um único arquivo ou todos os arquivos em um diretório. Um único arquivo utiliza uma thread. Exemplo: /user/hive/warehouse/mytable01/data.csv.

• Opção 2: Caminho com curinga — Lê múltiplos arquivos correspondentes a um padrão. O HDFS Reader aceita * (corresponde a quaisquer caracteres) e ? (corresponde a um único caractere). Exemplo: /hadoop/data_201704*. O número real de threads corresponde ao menor valor entre a quantidade de arquivos correspondentes e a configuração concurrent.

• Opção 3: Caminho de partição — Lê dados de um diretório de partição do Hive. Quando uma tabela Hive é criada com partições (por exemplo, partition(day="20150820", hour="09")), a partição aparece como uma estrutura de diretórios no HDFS. Para ler todos os dados de um determinado dia, defina o caminho da seguinte forma:

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

O Data Integration trata todos os arquivos em uma tarefa de sincronização como uma única tabela. Todos os arquivos devem estar em conformidade com o mesmo esquema, e a conta admin deve ter permissões de leitura sobre eles. Se os nomes dos arquivos seguirem um padrão baseado em tempo, use parâmetros de agendamento para substituir o caminho dinamicamente com base no horário comercial.

Notas sobre análise de formato de arquivo

TextFile e ORCFile analisam tipos complexos do Hive de maneira diferente. Para o tipo map, um ORCFile produz {job=80, team=60}, enquanto um TextFile produz {job:80, team:60}. Os dados são os mesmos, mas o formato difere. Se seus dados incluírem tipos complexos do Hive, use um formato de arquivo consistente em todo o caminho. Para unificar o formato, exporte tabelas TextFile para ORCFile no cliente Hive.

Formato de esquema Parquet

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

• MessageTypeName: Nome para o tipo de mensagem.

• RequiredStatus: Use required para colunas não nulas e optional para colunas anuláveis. Defina todas as colunas como optional.

• DataType: Os tipos compatíveis são BOOLEAN, INT32, INT64, INT96, FLOAT, DOUBLE, BINARY (use para tipos string) e FIXED_LEN_BYTE_ARRAY.

• Termine cada definição de coluna com ponto e vírgula, incluindo a última.

Exemplo:

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

Configuração do leitor CSV

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

Todos os campos disponíveis e seus padrões:

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;

Configuração de HA do 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"
}

Os parâmetros dfs.data.transfer.protection , dfs.datanode.use.datanode.hostname e dfs.client.use.datanode.hostname habilitam a autenticação Kerberos no plugin HDFS Reader. Se a autenticação Kerberos já estiver configurada na fonte de dados HDFS, esses parâmetros não são necessários na configuração do plugin. Consulte Configure uma fonte de dados HDFS .

Exemplo de configuração Kerberos

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

Como o Kerberos requer o caminho absoluto para o arquivo keytab, implante esta configuração em um grupo de recursos.

{
    "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 gravação

O HDFS Writer usa uma estratégia de gravação e renomeação: ele grava primeiro em uma pasta temporária e depois move os arquivos para o caminho de destino. O parâmetro writeMode controla como os arquivos existentes com o mesmo prefixo fileName são tratados antes do início da gravação.

O formato Parquet não aceita o modo append . Use nonConflict para arquivos Parquet.

Gravação em HDFS com base em OSS

Quando dataxParquetMode é fields, o HDFS Writer aceita OSS como armazenamento subjacente. Adicione os seguintes parâmetros OSS ao 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"
    }
}

Substitua os seguintes espaços reservados pelos valores reais: