本文将介绍使用Spark batch作业处理存储于Hive中Twitter数据的一些实用技巧。
首先我们需要引入一些依赖包，参考如下：
name := "Sentiment"version := "1.0"
scalaVersion := "2.10.6"
assemblyJarName in assembly := "sentiment.jar"
libraryDependencies += "org.apache.spark" % "spark-core_2.10" % "1.6.0" % "provided"
libraryDependencies += "org.apache.spark" % "spark-sql_2.10" % "1.6.0" % "provided"
libraryDependencies += "org.apache.spark" %% "spark-hive" % "1.6.0" % "provided"
libraryDependencies += "edu.stanford.nlp" % "stanford-corenlp" % "3.5.1"
libraryDependencies += "edu.stanford.nlp" % "stanford-corenlp" % "3.5.1" classifier "models"
resolvers += "Akka Repository" at "http://repo.akka.io/releases/"
assemblyMergeStrategy in assembly := {
case PathList("META-INF", xs @ _*) => MergeStrategy.discard
case x => MergeStrategy.first}
编写一个Scala case class用于存储解析好的Twitter Json数据：
case class Tweet(coordinates: String, geo:String, handle: String,
hashtags: String, language: String,
location: String, msg: String, time: String,
tweet_id: String, unixtime: String,
user_name: String, tag: String,
profile_image_url: String,
source: String, place: String, friends_count: String,
followers_count: String, retweet_count: String,
time_zone: String, sentiment: String,
stanfordSentiment: String)
引入以下的包：
import java.util.Properties
import com.vader.SentimentAnalyzer
import edu.stanford.nlp.ling.CoreAnnotations
import edu.stanford.nlp.neural.rnn.RNNCoreAnnotations
import edu.stanford.nlp.pipeline.StanfordCoreNLP
import edu.stanford.nlp.sentiment.SentimentCoreAnnotations
import org.apache.log4j.{Level, Logger}
import org.apache.spark.{<span class="wp_keywordlink_affiliate"><a data-original-title="View all posts in Spark" href="/archives/tag/spark" title="" target="_blank">Spark</a></span>Conf, <span class="wp_keywordlink_affiliate"><a data-original-title="View all posts in Spark" href="/archives/tag/spark" title="" target="_blank">Spark</a></span>Context}
import org.apache.spark.serializer.KryoSerializer
import org.apache.spark.sql._
用Scala编写的用于从Hive中读取数据的Spark代码片段：
def main(args: Array[String]) {
Logger.getLogger("org.apache.spark").setLevel(Level.ERROR)
Logger.getLogger("org.apache.spark.storage.BlockManager").setLevel(Level.ERROR)
val logger: Logger = Logger.getLogger("com.iteblog.sentiment.TwitterSentimentAnalysis")
val sparkConf = new SparkConf().setAppName("TwitterSentimentAnalysis")
sparkConf.set("spark.streaming.backpressure.enabled", "true")
sparkConf.set("spark.cores.max", "32")
sparkConf.set("spark.serializer", classOf[KryoSerializer].getName)
sparkConf.set("spark.sql.tungsten.enabled", "true")
sparkConf.set("spark.eventLog.enabled", "true")
sparkConf.set("spark.app.id", "Sentiment")
sparkConf.set("spark.io.compression.codec", "snappy")
sparkConf.set("spark.rdd.compress", "true")
val sc = new SparkContext(sparkConf)
val sqlContext = new org.apache.spark.sql.hive.HiveContext(sc)
import sqlContext.implicits._
val tweets = sqlContext.read.json("hdfs://www.iteblog.com:8020/social/twitter")
sqlContext.setConf("spark.sql.orc.filterPushdown", "true")
tweets.printSchema()
tweets.count
tweets.take(5).foreach(println)
其中我们需要注意的是我们需要创建Hive context而不是标准的SQL context
在运行我们的代码之前，先确认Hive中存储Twitter Json数据的表，以及用于存放结果数据的表格是否存在，本文用于存储结果数据的表格使用了ORC 格式
beeline
!connect jdbc:hive2://localhost:10000/default;
!set showHeader true;
set hive.vectorized.execution.enabled=true;
set hive.execution.engine=tez;
set hive.vectorized.execution.enabled =true;
set hive.vectorized.execution.reduce.enabled =true;
set hive.compute.query.using.stats=true;
set hive.cbo.enable=true;
set hive.stats.fetch.column.stats=true;
set hive.stats.fetch.partition.stats=true;
show tables;
describe sparktwitterorc;
describe twitterraw;
describe sparktwitterorc;
analyze table sparktwitterorc compute statistics;
analyze table sparktwitterorc compute statistics for columns;
上面名为twitterraw的表格是用于存放Twitter Json数据的表；而名为sparktwitterorc的表格是用于存放Spark处理结果的表。
如何将RDD或者DataFrame中的数据写入到Hive ORC表呢？操作如下：
outputTweets.toDF().write.format("orc").mode(SaveMode.Overwrite).saveAsTable("default.sparktwitterorc")
在编译的程序时候设置JVM相关参数
export SBT_OPTS="-Xmx2G -XX:+UseConcMarkSweepGC -XX:+CMSClassUnloadingEnabled -XX:MaxPermSize=2G -Xss2M -Duser.timezone=GMT"
sbt -J-Xmx4G -J-Xms4G assembly
将Spark作业提交到YARN集群：
spark-submit --class com.iteblog.sentiment.TwitterSentimentAnalysis --master yarn-client sentiment.jar --verbose
这里附上我们的rawtwitter表建表语句：
CREATE TABLE rawtwitter
(
handle STRING,
hashtags STRING,
msg STRING,
language STRING,
time STRING,
tweet_id STRING,
unixtime STRING,
user_name STRING,
geo STRING,
coordinates STRING,
`location` STRING,
time_zone STRING,
retweet_count STRING,
followers_count STRING,
friends_count STRING,
place STRING,
source STRING,
profile_image_url STRING,
tag STRING,
sentiment STRING,
stanfordsentiment STRING
)
ROW FORMAT SERDE 'org.apache.hive.hcatalog.data.JsonSerDe'
LOCATION 'hdfs://www.iteblog.com:8020/social/twitter'