Passer un nom de fichier à Java UDF à partir de Pig en utilisant le cache distribué

Question

Je suis en train d'utiliser un petit fichier de carte dans ma fonction UDF Java et je veux passer le nom de ce fichier depuis Pig à travers le constructeur.

Voici la partie pertinente de ma fonction UDF

public GenerateXML(String mapFilename) throws IOException {
    this(null);
}

public GenerateXML(String mapFilename) throws IOException {
    if (mapFilename != null) {
        // do preocessing
    }
}

Dans le script Pig, j'ai la ligne suivante

DEFINE GenerateXML com.domain.GenerateXML('typemap.tsv');

Cela fonctionne en mode local, mais pas en mode distribué. Je passe les paramètres suivants à Pig en ligne de commande

pig -Dmapred.cache.files="/path/to/typemap.tsv#typemap.tsv" -Dmapred.create.symlink=yes -f generate-xml.pig

Et je reçois l'exception suivante

2013-01-11 10:39:42,002 [main] ERROR org.apache.pig.tools.grunt.Grunt - ERROR 1200: Pig script failed to parse: 
 Failed to generate logical plan. Nested exception: java.lang.RuntimeException: could not instantiate 'com.domain.GenerateXML' with arguments '[typemap.tsv]'

Avez-vous une idée de ce que je dois changer pour que cela fonctionne ?

Answer 1

Le problème est maintenant résolu.

Il semble que lorsque j'exécute le script Pig en utilisant les paramètres suivants

pig -Dmapred.cache.files="/chemin/vers/typemap.tsv#typemap.tsv" -Dmapred.create.symlink=yes -f generate-xml.pig

Le /chemin/vers/typemap.tsv devrait être le chemin local et non un chemin dans HDFS.

Answer 2

Vous pouvez utiliser la fonction getCacheFiles dans une UDF Pig et cela suffira - vous n'avez pas besoin d'utiliser des propriétés supplémentaires comme mapred.cache.files. Votre cas peut être implémenté comme ceci :

public class UdfCacheExample extends EvalFunc {

    private Dictionary dictionary;
    private String pathToDictionary;

    public UdfCacheExample(String pathToDictionary) {
        this.pathToDictionary = pathToDictionary;
    }

    @Override
    public Tuple exec(Tuple input) throws IOException {
        Dictionary dictionary = getDictionary();
        return createSomething(input);
    }

    @Override
    public List getCacheFiles() {
        return Arrays.asList(pathToDictionary);
    }

    private Dictionary getDictionary() {
        // initialisation paresseuse ici
    }
}