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

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

简而言之，我想从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>

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

谢谢!

最佳答案

StroreFunc的构造函数在前端和后端都被调用。当从前端调用它时(在启 Action 业之前)，您将获得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));
      ...
  ...
}