Analyser CSV comme DataFrame/DataSet avec Apache Spark et Java

Procédure

• Créer une Classe (Schéma) pour encapsuler votre structure (il n'est pas nécessaire pour l'approche B, mais ce serait rendre votre code plus facile à lire si vous utilisez Java)

  public class Record implements Serializable {
    String department;
    String designation;
    long costToCompany;
    String state;
    //constructor , getters and setters  
  }

• Chargement CVS (JSON) fichier

  JavaSparkContext sc;
  JavaRDD<String> data = sc.textFile("path/input.csv");
  //JavaSQLContext sqlContext = new JavaSQLContext(sc); //For previous versions 
  SQLContext sqlContext = new SQLContext(sc); //In Spark 1.3 the Java API and Scala API have been unified
  
  
  JavaRDD<Record> rdd_records = sc.textFile(data).map(
    new Function<String, Record>() {
        public Record call(String line) throws Exception {
           //Here you can use JSON
           //Gson gson = new Gson();
           //gson.fromJson(line, Record.class);
           String[] fields = line.split(",");
           Record sd = new Record(fields[0], fields[1], fields[2].trim(), fields[3]);
           return sd;
        }
  });

À ce stade, vous avez 2 méthodes:

A. SparkSQL

• Enregistrer un tableau (à l'aide de votre Schéma défini Classe)

  JavaSchemaRDD table = sqlContext.applySchema(rdd_records, Record.class);
  table.registerAsTable("record_table");
  table.printSchema();

• Requête de la table avec votre Requête souhaitée-groupe-par

  JavaSchemaRDD res = sqlContext.sql("
    select department,designation,state,sum(costToCompany),count(*) 
    from record_table 
    group by department,designation,state
  ");

• Ici vous serait également en mesure de faire toute autre demande que vous désirez, à l'aide d'un SQL approche

B. Étincelle

• De cartographie à l'aide d'une clé composée: Department,Designation,State

  JavaPairRDD<String, Tuple2<Long, Integer>> records_JPRDD = 
  rdd_records.mapToPair(new
    PairFunction<Record, String, Tuple2<Long, Integer>>(){
      public Tuple2<String, Tuple2<Long, Integer>> call(Record record){
        Tuple2<String, Tuple2<Long, Integer>> t2 = 
        new Tuple2<String, Tuple2<Long,Integer>>(
          record.Department + record.Designation + record.State,
          new Tuple2<Long, Integer>(record.costToCompany,1)
        );
        return t2;
  }

  });

• reduceByKey à l'aide de la clé composite, résumant costToCompany colonne, et l'accumulation du nombre d'enregistrements par clé

  JavaPairRDD<String, Tuple2<Long, Integer>> final_rdd_records = 
   records_JPRDD.reduceByKey(new Function2<Tuple2<Long, Integer>, Tuple2<Long,
   Integer>, Tuple2<Long, Integer>>() {
      public Tuple2<Long, Integer> call(Tuple2<Long, Integer> v1,
      Tuple2<Long, Integer> v2) throws Exception {
          return new Tuple2<Long, Integer>(v1._1 + v2._1, v1._2+ v2._2);
      }
  });

OriginalL'auteur emecas

Suivantes pourraient ne pas être tout à fait correcte, mais elle devrait vous donner une idée de comment jongler avec les données. C'est pas joli, doit être remplacé par cas, classes, etc, mais comme un rapide exemple d'utilisation de l'étincelle de l'api, j'espère que c'est suffisant 🙂

val rawlines = sc.textfile("hdfs://.../*.csv")
case class Employee(dep: String, des: String, cost: Double, state: String)
val employees = rawlines
  .map(_.split(",") /*or use a proper CSV parser*/
  .map( Employee(row(0), row(1), row(2), row(3) )

# the 1 is the amount of employees (which is obviously 1 per line)
val keyVals = employees.map( em => (em.dep, em.des, em.state), (1 , em.cost))

val results = keyVals.reduceByKey{ a,b =>
    (a._1 + b._1, b._1, b._2) # (a.count + b.count , a.cost + b.cost )
}

#debug output
results.take(100).foreach(println)

results
  .map( keyval => someThingToFormatAsCsvStringOrWhatever )
  .saveAsTextFile("hdfs://.../results")

Ou vous pouvez utiliser SparkSQL:

val sqlContext = new SQLContext(sparkContext)

# case classes can easily be registered as tables
employees.registerAsTable("employees")

val results = sqlContext.sql("""select dep, des, state, sum(cost), count(*) 
  from employees 
  group by dep,des,state"""

OriginalL'auteur jkgeyti