0

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

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

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

谢谢！

4

1 回答 1

1

StroreFunc 的构造函数在frontend和backend都被调用。当从前端调用它时（在作业启动之前），您将获得 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)); 
      ...
  ...
}
于 2014-09-01T21:42:27.707 回答