hadoop - 在 Pig StoreFunc 中访问分布式缓存

Question

我已经查看了有关该主题的所有其他主题，但仍未找到答案...

简单地说，我想从 Pig StoreFunc 访问 hadoop 分布式缓存，而不是直接从 UDF 中访问。

相关 PIG 代码行:

DEFINE CustomStorage KeyValStorage('param1','param2','param3');
...
STORE BLAH INTO /path/ using CustomStorage();

相关 Java 代码:

public class KeyValStorage<M extends Message> extends BaseStoreFunc /* ElephantBird Storage which inherits from StoreFunc */ {

...
public KeyValStorage(String param1, String param2, String param3) {
    ...
        try {
            InputStream is = new FileInputStream(configName);
            try {
                prop.load(is);
            } catch (IOException e) {
                System.out.println("PROPERTY LOADING FAILED");
                e.printStackTrace();
            }
        } catch (FileNotFoundException e) {
            System.out.println("FILE NOT FOUND");
            e.printStackTrace();
        }
   }
...
}

configName 是我应该能够从分布式缓存中读取的本地文件的名称，但是，我收到了 FileNotFoundException。当我直接从 PIG UDF 中使用完全相同的代码时，会找到该文件，因此我知道该文件是通过分布式缓存传送的。我设置了适当的参数以确保发生这种情况:

<property><name>mapred.cache.files</name><value>/path/to/file/file.properties#configName</value></property>

有什么办法可以解决这个问题吗？

谢谢!

Answer 1

StroreFunc 的构造函数在前端 和后端 都被调用。当从前端调用它时(在作业启动之前)，您将得到 FileNotFoundException，因为此时分布式缓存中的文件尚未复制到节点的本地磁盘。
您可以检查您是否在后端(当作业正在执行时)并仅在这种情况下加载文件，例如:

DEFINE CustomStorage KeyValStorage('param1','param2','param3');
set mapreduce.job.cache.files hdfs://host/user/cache/file.txt#config
...
STORE BLAH INTO /path/ using CustomStorage();

public KeyValStorage(String param1, String param2, String param3) {
  ...
  try {
    if (!UDFContext.getUDFContext().isFrontend()) {
      InputStream is = new FileInputStream("./config");
      BufferedReader br = new BufferedReader(new InputStreamReader(is)); 
      ...
  ...
}