Ecriture de fichiers dans HDFS à partir de fonctions de transformation d'étincelles dans Java

Question

Ma question est similaire à une question déjà posée qui se trouve dans scala et pour la lecture de fichiers.

Reading files dynamically from HDFS from within spark transformation functions

Je suis au courant des accumulateurs qui les utilisent pour retourner des résultats au conducteur et à l'écriture HDFS dans le pilote. Dans notre cas d'utilisation, la sortie de chaque exécuteur est grande, donc je cherche un moyen d'écrire dans HDFS dans les transformations en Java.

Merci!

Answer 1

Enfin trouvé un moyen gracieux pour y parvenir. Créez une variable de diffusion pour la configuration Hadoop

Configuration configuration = JavaSparkContext.toSparkContext(context).hadoopConfiguration(); 
Broadcast<SerializableWritable<Configuration>> bc = context.broadcast(new SerializableWritable<Configuration>(configuration)); 

passe cette variable de diffusion à votre transformation ou de l'action et obtenir système de fichiers Hadoop en utilisant extrait de code suivant:

FileSystem fileSystem = FileSystem.get(bc.getValue().value()); 

En espérant que cela aide si quelqu'un d'autre dans le même bateau.

À la votre!

Answer 2

JavaPairInputDStream<String, byte[]> input = KafkaUtils.createJDQDirectStream(ssc, String.class, byte[].class, 
     StringDecoder.class, DefaultDecoder.class, kafkaParams, Collections.singleton(topicName)); 

JavaPairDStream<String, byte[]> output = input.transformToPair(new Function<JavaPairRDD<String, byte[]>, JavaPairRDD<String, byte[]>>() { 
    public JavaPairRDD<String, byte[]> call(JavaPairRDD<String, byte[]> stringJavaPairRDD) throws Exception { 
     JavaSparkContext sc = JavaSparkContext.fromSparkContext(stringJavaPairRDD.context()); 
     stringJavaPairRDD.saveAsTextFile("hdfs://"); 
     return stringJavaPairRDD; 
    } 
});