从多个服务器加载数据时避免数据重复

Question

我有十几个 Web 服务器，每个服务器都将数据写入日志文件。每小时开始时，使用运行命令的 cron 脚本将前一小时的数据加载到 hive：

hive -e "LOAD DATA LOCAL INPATH 'myfile.log' INTO TABLE my_table PARTITION(dt='2015-08-17-05')"

在某些情况下，命令会失败并以非 0 的代码退出，在这种情况下，我们的脚本会等待并重试。问题是，在某些失败的情况下，数据加载不会失败，即使它显示了失败消息。如何确定数据是否已加载？

加载数据的“失败”示例：

加载数据到表 default.my_table 分区 (dt=2015-08-17-05) 异常失败 org.apache.hadoop.hive.ql.metadata.HiveException：无法更改 分割。失败：执行错误，返回代码 1 从 org.apache.hadoop.hive.ql.exec.MoveTask

编辑： 或者，有没有办法查询 hive 中加载的文件名？我可以使用DESCRIBE 查看文件数量。我能知道他们的名字吗？

Answer 1

关于“分区中加载了哪些文件”：

• 如果您使用了EXTERNAL TABLE 并且刚刚上传了您的原始数据 HDFS 目录中的文件映射到LOCATION，然后你可以

(a) 只需从命令行在该目录上运行 hdfs dfs -ls（或使用等效的 Java API 调用） (b) 运行 Hive 查询，例如 select distinct INPUT__FILE__NAME from (...)

• 但在您的情况下，您将数据复制到“托管”表中，因此 无法检索数据沿袭（即使用了哪个日志文件 创建每个托管数据文件）
• ...除非您在日志文件中显式添加原始文件名inside， 当然（在“特殊”标题记录上，或在每条记录的开头 - 这可以用旧的 sed 完成）

---

关于“如何自动避免 INSERT 上的重复”：有一种方法，但需要进行相当多的重新设计，并且会花费您处理时间/（额外的 Map 步骤加上 MapJoin）/。 ..

1. 将您的日志文件映射到EXTERNAL TABLE，以便您可以运行 INSERT-SELECT 查询
2. 使用INPUT__FILE__NAME 伪列作为源将原始文件名上传到托管表中

3. 添加一个带有相关子查询的WHERE NOT EXISTS 子句，这样如果源文件名已经存在于目标中，那么你就不再加载了

   INSERT INTO TABLE Target SELECT ColA, ColB, ColC, INPUT__FILE__NAME AS SrcFileName FROM Source src WHERE NOT EXISTS (SELECT DISTINCT 1 FROM Target trg WHERE trg.SrcFileName =src.INPUT__FILE__NAME )

   注意实际上需要的愚蠢的 DISTINCT 以避免炸毁 Mappers 中的 RAM；对于像 Oracle 这样成熟的 DBMS 来说，这将毫无用处，但 Hive 优化器仍然相当粗糙......

Answer 2

我不相信你可以在 Hadoop/Hive 中简单地做到这一点。所以这里是python实现的基础：

import subprocess
x=subprocess.check_output([hive -e "select count(*) from my_table where dt='2015-08-17-05'"])
print type(x)
print x

但是您必须花一些时间处理反斜杠才能让 hive -e 使用 python 工作。这可能非常困难。首先编写一个包含该简单查询的文件，然后使用hive -f filename 可能更容易。然后，打印subprocess.check_output 的输出，以查看输出是如何存储的。您可能需要进行一些正则表达式或类型转换，但我认为它应该以字符串的形式返回。然后只需使用 if 语句：

if x > 0:
    pass
else:
    hive -e "LOAD DATA LOCAL INPATH 'myfile.log' INTO TABLE my_table PARTITION(dt='2015-08-17-05')"