jak zebrać wynik iskry sql do pliku?

Question

Poniżej znajduje się mój iskrowy skrypt sql, który ładuje plik i używa na nim kodu SQL, chcę zebrać dane wyjściowe z kwerendy sql i zapisać je w pliku, nie wiem, jak komukolwiek pomóc.

//import classes for sql 
 
import org.apache.spark.sql.SQLContext 
 
import org.apache.spark.{SparkConf, SparkContext} 
 

 
val sqlContext = new org.apache.spark.sql.SQLContext(sc) 
 

 
// createSchemaRDD is used to implicitly convert an RDD to a SchemaRDD. 
 
import sqlContext.createSchemaRDD 
 

 

 
//hdfs paths 
 
val warehouse="hdfs://quickstart.cloudera/user/hive/warehouse/" 
 
val customers_path=warehouse+"people/people.txt" 
 
customers_path 
 

 
//create rdd file called file 
 
val file=sc.textFile(customers_path) 
 

 
val schemaString="name age" 
 

 
import org.apache.spark.sql._ 
 

 

 

 
val schema = 
 
    StructType(
 
    schemaString.split(",").map(fieldName => StructField(fieldName, StringType, true))) 
 

 
val rowRDD=file.map(_.split(",")).map(p => Row(p(0),p(1).trim)) 
 

 
val peopleSchemRDD=sqlContext.applySchema(rowRDD, schema) 
 

 
// Register the SchemaRDD as a table. 
 
peopleSchemRDD.registerTempTable("people") 
 

 
// SQL statements can be run by using the sql methods provided by sqlContext. 
 
sqlContext.sql("select count(*) from people").collect().foreach(println) 
 
System.exit(0)

Answer 1

Jeśli chcesz po prostu policzyć liczbę wierszy w dużym pliku na HDFS i zapisać go do innego pliku:

import java.nio.file.{ Files, Paths } 
val path = "hdfs://quickstart.cloudera/user/hive/warehouse/people/people.txt" 
val rdd = sc.textFile(path) 
val linesCount = rdd.count 
Files.write(Paths.get("line_count.txt"), linesCount.toString.getBytes) 

Answer 2

//import classes for sql 
 
import sqlContext.implicits._ 
 
import org.apache.spark.sql.SQLContext 
 
import org.apache.spark.{SparkConf, SparkContext} 
 

 
val sqlContext = new org.apache.spark.sql.SQLContext(sc) 
 

 
// createSchemaRDD is used to implicitly convert an RDD to a SchemaRDD. 
 
import sqlContext.createSchemaRDD 
 
import sqlContext.implicits._ 
 

 
//hdfs paths 
 
val warehouse="hdfs://quickstart.cloudera/user/hive/warehouse/" 
 
val customers_path=warehouse+"people/people.txt" 
 
customers_path 
 

 
//create rdd file called file 
 
val file=sc.textFile(customers_path) 
 

 
val schemaString="name age" 
 

 
import org.apache.spark.sql._ 
 

 

 

 
val schema = 
 
    StructType(
 
    schemaString.split(",").map(fieldName => StructField(fieldName, StringType, true))) 
 

 
val rowRDD=file.map(_.split(",")).map(p => Row(p(0),p(1).trim)) 
 

 
val peopleSchemRDD=sqlContext.applySchema(rowRDD, schema) 
 

 
// Register the SchemaRDD as a table. 
 
peopleSchemRDD.registerTempTable("people") 
 

 
// SQL statements can be run by using the sql methods provided by sqlContext. 
 
val op=sqlContext.sql("select count(*) from people") 
 
val c=op.collect() 
 
val rdd=sc.parallelize(c) 
 
rdd.saveAsTextFile("/home/cloudera/op") 
 
System.exit(0)