RDDs

    编程指定模式

    • 从原来的RDD创建一个行的RDD
    • 创建由一个表示的模式与第一步创建的RDD的行结构相匹配
    • 在行RDD上通过applySchema方法应用模式
    1. // sc is an existing SparkContext.
    2. val sqlContext = new org.apache.spark.sql.SQLContext(sc)
    4. // Create an RDD
    5. val people = sc.textFile("examples/src/main/resources/people.txt")
    7. val schemaString = "name age"
    10. import org.apache.spark.sql._
    12. // Generate the schema based on the string of schema
    13. val schema =
    14.   StructType(
    15.     schemaString.split(" ").map(fieldName => StructField(fieldName, StringType, true)))
    17. // Convert records of the RDD (people) to Rows.
    20. val peopleSchemaRDD = sqlContext.applySchema(rowRDD, schema)
    22. // Register the SchemaRDD as a table.
    23. peopleSchemaRDD.registerTempTable("people")
    25. // SQL statements can be run by using the sql methods provided by sqlContext.
    26. val results = sqlContext.sql("SELECT name FROM people")
    28. // The results of SQL queries are SchemaRDDs and support all the normal RDD operations.