如何将sparksql输出收集到一个文件中

Question

下面是我的spark sql脚本，它加载一个文件并在上面使用sql，我想收集sql查询的输出并将其写入一个文件，不知道如何帮助任何人。

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

如果您只想计算HDFS上一个大文件中的行数，并将其写入另一个文件：

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)

Answer 3

peopleSchemaRDD.registerTempTable("people")
val op=sqlContext.sql("select * from people").count().toString
val pw=new PrintWriter(new File("path"))
pw.write("count of people:"+op+"n")
pw.close()

创建一个名为people的临时表，然后编写一个查询以获得所需的输出和count函数，该函数计算输出转换为tostring后的行数。这个存储在object op中的值是使用print编写器调用的，以便将其写入文本文件中。如果people列包含重复的值，则使用distinct关键字来区分sql查询中的唯一值。