在 hadoop 中操作行数据以添加缺失的列答案

【问题标题】：Manipulate row data in hadoop to add missing columns在 hadoop 中操作行数据以添加缺失的列
【发布时间】：2013-08-20 19:12:35
【问题描述】：

我有来自 IIS 的日志文件存储在 hdfs 中，但由于网络服务器配置的原因，一些日志没有包含所有列，或者它们以不同的顺序出现。我想生成具有通用架构的文件，以便可以在它们之上定义 Hive 表。

好的日志示例：

#Fields: date time s-ip cs-method cs-uri-stem useragent
2013-07-16 00:00:00 10.1.15.8 GET /common/viewFile/1232 Mozilla/5.0+Chrome/27.0.1453.116

缺少列的示例日志（缺少cs-method和useragent）：

#Fields: date time s-ip cs-uri-stem 
2013-07-16 00:00:00 10.1.15.8 /common/viewFile/1232

缺少列的日志需要像这样映射到完整架构：

#Fields: date time s-ip cs-method cs-uri-stem useragent
2013-07-16 00:00:00 10.1.15.8 null /common/viewFile/1232 null

不良日志可以启用任意列组合，并且以不同的顺序排列。

如何根据日志文件中的字段行将可用列映射到完整架构？

编辑：通常我会通过将列模式定义为将列名映射到索引的字典来解决此问题。即： col['date']=0 col['time']=1 等然后我会从文件中读取#Fields行并解析出启用的列并生成标题dict映射标题名称到文件中的列索引.然后对于剩余的数据行，我通过索引知道它的标题，通过 header=column name 将其映射到我的列模式，并以正确的顺序生成新行，插入带有空数据的缺失列。我的问题是我不明白如何在 hadoop 中执行此操作，因为每个地图都是单独执行的，因此如何与每个地图共享 #Fields 信息？

【问题讨论】：

这不是真正的猪问题。告诉我们你将如何用任意语言解决这个问题，包括伪代码。如果给你一个任意字段，你将如何决定它的含义以及它应该放在哪一列？
我添加了说明，如果我可以在处理每个文件时从每个文件中提取和共享字段信息，我将如何做到这一点。否则我认为 RegexSerDe 应该能够映射字段，但是当您在文件中有可用数据时，这不是浪费吗？

标签： hadoop hive apache-pig

【解决方案1】：

您可以使用this 将标题应用于创建地图的列。从那里你可以使用像这样的 UDF：

myudf.py

#!/usr/bin/python

@outputSchema('newM:map[]')
def completemap(M):
    if M is None:
        return None
    to_add = ['A', 'D', 'F']
    for item in to_add:
        if item not in M:
            M[item] = None
    return M

@outputSchema('A:chararray, B:chararray, C:chararray, D:chararray, E:chararray, F:chararray')
def completemap_v2(M):
    if M is None:
        return (None,
                None,
                None,
                None,
                None,
                None)
    return (M.get('A', None),
            M.get('B', None),
            M.get('C', None),
            M.get('D', None),
            M.get('E', None),
            M.get('F', None))

将缺失的元组添加到地图中。

示例输入：

csv1.in             csv2.in
-------            ---------
A|B|C               D|E|F
Hello|This|is       PLEASE|WORK|FOO
FOO|BAR|BING        OR|EVERYTHING|WILL
BANG|BOSH           BE|FOR|NAUGHT

示例脚本：

A = LOAD 'tests/csv' USING myudfs.ExampleCSVLoader('\\|') AS (M:map[]); 
B = FOREACH A GENERATE FLATTEN(myudf.completemap_v2(M));

输出：

B: {null::A: chararray,null::B: chararray,null::C: chararray,null::D: chararray,null::E: chararray,null::F: chararray}
(,,,,,)
(,,,PLEASE,WORK,FOO)
(,,,OR,EVERYTHING,WILL)
(,,,BE,FOR,NAUGHT)
(,,,,,)
(Hello,This,is,,,)
(FOO,BAR,BING,,,)
(BANG,BOSH,,,,)

【讨论】：