Spark Sql系统入门4：spark应用程序中使用spark sql
Spark Sql系统入门4：spark应用程序中使用spark sql
问题导读 1.你认为如何初始化spark sql？ 2.不同的语言，实现方式都是什么？ 3.spark sql语句如何实现在应用程序中使用？ 为了使用spark sql，我们构建HiveContext （或则SQLContext 那些想要的精简版）基于我们的SparkContext.这个context 提供额外的函数为查询和整合spark sql数据。使用HiveContext，我们构建SchemaRDDs.这代表我们机构化数据，和操作他们使用sql或则正常的rdd操作如map（）.
初始化spark sql 为了开始spark sql，我们需要添加一些imports 到我们程序。如下面例子1 例子1Scala SQL imports
[Scala] 纯文本查看 复制代码
// Import Spark SQL
import org.apache.spark.sql.hive.HiveContext
// Or if you can't have the hive dependencies
import org.apache.spark.sql.SQLContextScala用户注意，我们不使用 import HiveContext._,像我们这样做SparkContext，获取访问implicits.这些implicits用来转换rdds,带着需要的type信息到spark sql的序列化rdds为查询。相反，一旦我们有了结构化HiveContext实例化，我们可以导入 implicits 在例子2中。导入Java和Python在例子3和4中。例子2Scala SQL imports
[Scala] 纯文本查看 复制代码
// Create a Spark SQL HiveContext
val hiveCtx = ...
// Import the implicit conversions
import hiveCtx._例子3Java SQL imports
[Java] 纯文本查看 复制代码
// Import Spark SQL
import org.apache.spark.sql.hive.HiveContext;
// Or if you can't have the hive dependencies
import org.apache.spark.sql.SQLContext;
// Import the JavaSchemaRDD
import org.apache.spark.sql.SchemaRDD;
import org.apache.spark.sql.Row;例子4Python SQL imports
[Python] 纯文本查看 复制代码
# Import Spark SQL
from pyspark.sql import HiveContext, Row
# Or if you can't include the hive requirements
from pyspark.sql import SQLContext, Row一旦我们添加我们的imports,我们需要创建HiveContext,或则SQLContext，如果我们引入Hive依赖（查看例子5和6）。这两个类都需要运行spark。 例子5：使用Scala结构化sql context
[Scala] 纯文本查看 复制代码
val sc = new SparkContext(...)
val hiveCtx = new HiveContext(sc)例子6：使用java结构化sql context
[Java] 纯文本查看 复制代码
JavaSparkContext ctx =
new JavaSparkContext(...);
SQLContext sqlCtx =
new HiveContext(ctx);例子7：使用python结构化sql context
[Python] 纯文本查看 复制代码
hiveCtx = HiveContext(sc)现在我们有了HiveContext 或则SQLContext，我们准备加载数据和查询。 基本查询例子 为了对一个表查询，我们调用HiveContext或则SQLContext的sql()函数.第一个事情，我们需要告诉spark sql关于一些数据的查询。在这种情况下，我们load Twitter数据【json格式】,和给它一个name,注册为 “临时表”，因此我们可以使用sql查询。 例子8使用Scala加载和查询tweets
[Scala] 纯文本查看 复制代码
val input = hiveCtx.jsonFile(inputFile)
// Register the input schema RDD
input.registerTempTable("tweets")
// Select tweets based on the retweetCount
val topTweets = hiveCtx.sql("SELECT text,
retweetCount FROM
tweets ORDER BY retweetCount LIMIT 10")例子9使用Java加载和查询tweets
[Java] 纯文本查看 复制代码
SchemaRDD input = hiveCtx.
jsonFile(inputFile);
// Register the input schema RDD
input.registerTempTable("tweets");
// Select tweets based on the retweetCount
SchemaRDD topTweets = hiveCtx.
sql("SELECT text,
retweetCount FROM
tweets ORDER BY retweetCount LIMIT 10");例子10使用Python加载和查询tweets
[Python] 纯文本查看 复制代码
input = hiveCtx.jsonFile(inputFile)
# Register the input schema RDD
input.registerTempTable("tweets")
# Select tweets based on the retweetCount
topTweets = hiveCtx.sql("""SELECT text,
retweetCount FROM
tweets ORDER BY retweetCount LIMIT 10""")如果你已经安装hive，并且复制hive-site.xml文件到$SPARK_HOME/conf，你也可以运行hiveCtx.sql 查询已存在的hive表。