le fichier CSV peut être divisé avec le lecteur CSV intégré Spark. Il sera de retour DataFrame / ensemble de données sur la lecture réussie du fichier. Sur le dessus de DataFrame/ DataSet, vous appliquez des opérations de type SQL facilement.

Utilisant Spark 2.x (et au-dessus) avec Java

Create SparkSession object alias spark

import org.apache.spark.sql.SparkSession;

SparkSession spark = SparkSession
    .builder()
    .appName("Java Spark SQL Example")
    .getOrCreate();

Créer un Schéma pour la Ligne avec StructType

import org.apache.spark.sql.types.StructType;

StructType schema = new StructType()
    .add("department", "string")
    .add("designation", "string")
    .add("ctc", "long")
    .add("state", "string");

créer une base de données à partir du fichier CSV et appliquer schéma

Dataset<Row> df = spark.read()
    .option("mode", "DROPMALFORMED")
    .schema(schema)
    .csv("path/input.csv");

plus d'options sur la lecture des données du fichier CSV

nous pouvons maintenant agréger les données de deux façons

1. Chemin SQL

enregistrer une table dans spark sql metastore pour effectuer L'opération SQL

df.createOrReplaceTempView("employee");

Exécuter une requête SQL sur les dataframe

Dataset<Row> sqlResult = spark.sql(
    "SELECT department, designation, state, SUM(ctc), COUNT(department)" 
        + " FROM employee GROUP BY department, designation, state");

sqlResult.show(); //for testing

nous pouvons même exécuter SQL directement sur le fichier CSV sans créer de table avec Spark SQL

2. Enchaînement d'objets ou programmation ou Java-like way

faire l'importation nécessaire pour les fonctions sql

import static org.apache.spark.sql.functions.count;
import static org.apache.spark.sql.functions.sum;

Utiliser groupBy et agg sur dataframe/dataset à effectuer count et sum sur les données

Dataset<Row> dfResult = df.groupBy("department", "designation", "state")
    .agg(sum("ctc"), count("department"));
// After Spark 1.6 columns mentioned in group by will be added to result by default

dfResult.show();//for testing

bibliothèques dépendantes

"org.apache.spark" % "spark-core_2.11" % "2.0.0" 
"org.apache.spark" % "spark-sql_2.11" % "2.0.0"

ce qui suit n'est peut-être pas tout à fait exact, mais cela devrait vous donner une idée de la façon de jongler avec les données. Il n'est pas joli, devrait être remplacé par des classes de cas, etc, mais comme un exemple rapide de la façon d'utiliser l'api spark, j'espère que c'est assez :)

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")

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"""

pour JSON, si votre fichier texte contient un objet JSON par ligne, vous pouvez utiliser sqlContext.jsonFile(path) pour Spark SQL charger un SchemaRDD (le schéma sera automatiquement déduite). Ensuite, vous pouvez l'enregistrer comme une table et des requêtes SQL. Vous pouvez également charger manuellement le fichier texte comme un RDD[String] contenant un objet JSON par enregistrement et utilisation sqlContext.jsonRDD(rdd) pour l'activer comme un SchemaRDD. jsonRDD est utile lorsque vous avez besoin de pré-traitement de vos données.