如何从行为类似于标准管道的两个命令中创建一个 linux shell 命令？

Question

我正在尝试使用 Hadoop Streaming 来运行两个命令，例如 gunzip | map_to_old_format.py，但是 gzip 错误提示“|.gz not found”或类似的内容（仅当通过 Hadoop 运行时.. 如果我运行在命令行上，它工作正常）。

由于我不知道如何即时在 Python 中进行 gunzip，因此我想创建一个 shell 脚本来为我执行此命令的组合（例如 gunzip_and_map_to_old.sh）。我尝试了以下方法，但 gzip 不喜欢（gzip 抱怨“gzip：stdin：不是 gzip 格式”）：

#!/bin/bash
while read data; do
    echo $data | gunzip | map_to_old_format.py $2
done

关于python gunzip，我尝试了f = gzip.GzipFile("", "rb", fileobj=sys.stdin) 以及here 描述的Wrapper 方法。

Answer 1

我对 Hadoop 一无所知，但我猜echo $data | gunzip 不起作用，因为$data 是data 的一行，而$data 本身可能不是gzip 格式。与其逐行传递数据，不如在 bash 脚本文件中这样做？

#!/bin/bash
gunzip | map_to_old_format.py

然后您可以通过像这样传入 gzip 文件来调用它：

cat data.gz | gunzip_and_map_to_old.sh

Answer 2

Hadoop 流通常使用 TextInputFormat 读取输入文件，并通过 std in 将其逐行传递给您的 python 映射器（使用制表符分隔键和值（在大多数情况下是行号和行文本）。

如果输入文件的文件扩展名不以 .gz 结尾，那么 hadoops TextInputFormat 在一次传递一行之前将不知道 gunzip 文件内容。您可以按照其他答案中的建议配置属性以强制 hadoop 压缩文件。

Answer 3

这并不能回答我的确切问题，但我可以通过在我的 Hadoop 命令 (source where I learned this) 中添加 -jobconf stream.recordreader.compression=gzip 来绕过它：

hadoop jar /usr/lib/hadoop/contrib/streaming/hadoop-streaming-*.jar \
    -jobconf stream.recordreader.compression=gzip \
    -D mapred.reduce.tasks=0 \
    -file map_to_old_format.py \
    -mapper map_to_old_format.py \
    -input /mydata/* -output output/newdata

注意：我仍然很好奇如何通过 shell 脚本完成上述操作，所以如果可能的话，请告诉我。