La conversion de JavaRDD à DataFrame Spark java

Importations:

import java.io.Serializable;

import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;

Créer une classe POJO pour les URL. Je vous recommande d'écrire pour le Journal de la ligne qui se compose d'url, la date, l'heure, la méthode, la cible, etc.. en tant que membres

public static class Url implements Serializable {
  private String value;

  public String getValue() {
    return value;
  }

  public void setValue(String value) {
    this.value = value;
  }
}  

Créer un EDR Url d'objets à partir d'un fichier texte

JavaRDD<Url> urlsRDD = spark.read()
  .textFile("/Users/karuturi/Downloads/log.txt")
  .javaRDD()
  .map(new Function<String, Url>() {
    @Override
    public Url call(String line) throws Exception {
      String[] parts = line.split("\\t");
      Url url = new Url();
      url.setValue(parts[0].replaceAll("[", ""));
      return url;
    }
  });

Créer DataFrame de RDD

Dataset<Row> urlsDF = spark.createDataFrame(urlsRDD, Url.class);

CA à DataFrame - Spark 2.0

CA à DataFrame - Spark 1.6

OriginalL'auteur mrsrinivas

Vous pouvez faire quelque chose comme (je suis la conversion à la volée à partir de la scala, donc excusez les fautes de frappe):

import org.apache.spark.sql.Row
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;

JavaRDD<Row> rowRDD = urlrdd.map(new Function<String, Row>() {
    @Override
    public Row call(String record) throws Exception {
        return RowFactory.create(record());
    }
}
//now you wish to create the target schema. This is basically a list of
//fields (each field would be a column) which you are adding to a StructType
List<StructField> fields = new ArrayList<>();
StructField field = DataTypes.createStructField("url", DataTypes.StringType, true);
fields.add(field);
StructType schema = DataTypes.createStructType(fields);

//now you can create the dataframe:
DataFrame df= sqlContext.createDataFrame(rowRDD, schema);    

Quelques remarques supplémentaires:

• Pourquoi êtes-vous flatmaping lorsque vous prenez le premier élément? Vous pourriez avoir tout simplement fait:

  JavaRDD<String> urlrdd=diskfile.flatMap(line -> line.split("\t")[0]);

• Je suppose que dans la vraie vie, vous voulez supprimer le '[' partir de l'url (vous pouvez facilement le faire à la carte).

• Si vous vous déplacez à spark 2.0 ou plus, alors au lieu de sqlContext vous devriez être en utilisant étincelle session (étincelle).

• Vous pouvez créer un dataframe avec toutes les colonnes. Vous pouvez le faire en ajoutant tous les champs dans le schéma (c'est à dire au lieu de faire juste un seul ajouter des champs de tous les ajouter). Au lieu d'utiliser urlrdd, utilisez diskfile et de faire la distinction à l'intérieur de la "public Ligne d'appel" de la création. Ce serait quelque chose comme ceci:

  JavaRDD<Row> rowRDD = diskfile.map(new Function<String, Row>() {
@override public Row call(String record) throws Exception {
String[] recs = record.split("\t")
return RowFactory.create(recs[0], recs[1], ...);
}
});

• Vous pouvez le créer directement: il suffit d'utiliser

  sqlContext.read.option("sep","\t").csv.load(filename,schema)

OriginalL'auteur Assaf Mendelson