Apache Spark，Cassandra和《权力的游戏》

Apache Spark和Cassandra是数据处理管道中的强大组合。在本文中，我们将使用Spark Cassandra组合构建一个Scala应用程序，并从《权力的游戏》中查询战斗数据。现在，我们将不做任何演出预测！但是，我们将展示攻击性最强的国王以及遭受攻击最多的国王。所以，您已经准备好了。

总览

我们的主要重点是Spark Cassandra与Scala集成的技术亮点。为此，我们将使用《权力的游戏》战斗数据加载Cassandra，然后使用Scala从Spark查询它。我们将在外壳程序中使用Spark，并将Spark Driver程序部署到群集中。我们将借助DataStax连接器以及使用SparkSQL生成DataFrame的方式来整理Scala案例类的示例。我们也混入了一些sbt配置。

在本文的“资源”部分的底部有截屏视频和相关链接。

此Spark Cassandra教程的目标读者是对Scala和Apache Spark具有中级经验的人。如果您想快速有效地达到此水平，请查看我们的Scala随需培训Apache Scala培训课程。

先决条件

Apache Cassandra（请参阅下面的资源）
下载的《权力的游戏》数据（请参阅下面的资源）
阿帕奇火花

大纲

将数据导入Cassandra
编写Scala代码
在SBT Shell中测试Spark Cassandra代码
使用SBT和spark-submit Spark Cassandra部署到Spark Cluster

Spark Cassandra示例

第1部分：准备Cassandra

让我们将GOT战斗数据导入Cassandra。为简单起见，我将使用本地运行的Cassandra实例。我在Mac上使用bin / cassandra脚本启动了Cassandra。（在Windows上使用cassandra.bat，但您已经知道这一点。）。

接下来，使用cqlsh连接到Cassandra并创建要使用的键空间。本教程创建一个“ gameofthrones”键空间：

CREATE KEYSPACE gameofthrones WITH replication = {'class': 'SimpleStrategy', 'replication_factor' : 1};

在这里，我们为战斗数据创建一个表格。

CREATE KEYSPACE gameofthrones WITH replication = {'class': 'SimpleStrategy', 'replication_factor' : 1};
CREATE TABLE gameofthrones.battles (
name TEXT,
year INT,
battle_number INT,
attacker_king TEXT,
defender_king TEXT,
attacker_1 TEXT,
attacker_2 TEXT,
attacker_3 TEXT,
attacker_4 TEXT,
defender_1 TEXT,
defender_2 TEXT,
defender_3 TEXT,
defender_4 TEXT,
attacker_outcome TEXT,
battle_type TEXT,
major_death TEXT,
major_capture TEXT,
attacker_size TEXT,
defender_size TEXT,
attacker_commander TEXT,
defender_commander TEXT,
summer TEXT,
location TEXT,
region TEXT,
note TEXT,
PRIMARY KEY(battle_number)
);

然后使用下面显示的Cassandra COPY导入战斗数据。（有关下载数据的位置，请参见下面的资源部分）。
顺便说一句，我需要运行Perl脚本，以使用perl -pi -e 's/\r/\n/g在CSV文件上将Mac的行尾编码更新为Unix。你的旅费可能会改变。

COPY battles (
name,
year,
battle_number,
attacker_king,
defender_king,
attacker_1,
attacker_2,
attacker_3,
attacker_4,
defender_1,
defender_2,
defender_3,
defender_4,
attacker_outcome,
battle_type,
major_death,
major_capture,
attacker_size,
defender_size,
attacker_commander,
defender_commander,
summer,
location,
region,
note)
FROM 'battles.csv'  // update this location as necessary
WITH HEADER = true;

内容包括第1部分。让我们继续第2部分，在其中编写一些Scala代码。

第2部分：Spark Cassandra Scala代码

（注意：以下所有示例代码（如果可以从Github获得）。下面的参考资料部分中的链接。）

首先，让我们布置项目的骨架结构–

mkdir got-battles // name it anything you'd likecd got-battles  // if you named it got-battlesmkdir srcmkdir src/mainmkdir src/main/scalamkdir src/main/scala/commkdir src/main/scala/com/supergloomkdir project

接下来，我们将为sbt和sbt-assembly插件添加一些文件。
首先是sbt的构建文件
got-battles / build.sbt文件：

name := "spark-cassandra-example"version := "1.0"assemblyOption in assembly := (assemblyOption in assembly).value.copy(includeScala = false)// https://groups.google.com/a/lists.datastax.com/forum/#!topic/spark-connector-user/5muNwRaCJnU
assemblyMergeStrategy in assembly <<= (assemblyMergeStrategy in assembly) {(old) => {case PathList("META-INF", "io.netty.versions.properties") => MergeStrategy.lastcase x => old(x)}
}scalaVersion := "2.10.6"resolvers += "jitpack" at "https://jitpack.io"libraryDependencies ++= Seq(
// use provided line when building assembly jar
// "org.apache.spark" %% "spark-sql" % "1.6.1" % "provided",
// comment above line and uncomment the following to run in sbt"org.apache.spark" %% "spark-sql" % "1.6.1","com.datastax.spark" %% "spark-cassandra-connector" % "1.5.0"
)

和第一行的got-battles / project / assembly.sbt文件：

addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.12.0")

现在让我们在get-battles / src / main / scala / com / supergloo中创建名为SparkCassandra.scala的Spark驱动程序代码

package com.superglooimport org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.functions._import com.datastax.spark.connector._
import com.datastax.spark.connector.rdd._
import org.apache.spark.sql.cassandra._/*** Simple Spark Cassandra * One example with Scala case class marshalling* Another example using Spark SQL */
object SparkCassandra {case class Battle(    battle_number: Integer,year: Integer,attacker_king: String,defender_king: String)def main(args: Array[String]) {val conf = new SparkConf().setAppName("SparkCassandraExampleApp")if (args.length > 0) conf.setMaster(args(0)) // in case we're running in sbt shell such as: run local[5]conf.set("spark.cassandra.connection.host", "127.0.0.1")  // so yes, I'm assuming Cassandra is running locally here.// adjust as neededval sc = new SparkContext(conf)// Spark Cassandra Example one which marshalls to Scala case classesval battles:Array[Battle] = sc.cassandraTable[Battle]("gameofthrones", "battles").select("battle_number","year","attacker_king","defender_king").toArraybattles.foreach { b: Battle =>println("Battle Number %s was defended by %s.".format(b.battle_number, b.defender_king))}// Spark Cassandra Example Two - Create DataFrame from Spark SQL readval sqlContext = new SQLContext(sc)val df = sqlContext.read.format("org.apache.spark.sql.cassandra").options(Map( "table" -> "battles", "keyspace" -> "gameofthrones" )).load()df.show// Game of Thrones Battle analysis // Who were the most aggressive kings?  (most attacker_kings)val countsByAttack = df.groupBy("attacker_king").count().sort(desc("count"))countsByAttack.show()// Which kings were attacked the most?  (most defender_kings)val countsByDefend = df.groupBy("defender_king").count().sort(desc("count"))countsByDefend.show()sc.stop()}
}

第3部分：从SBT控制台运行Spark Cassandra Scala代码

通过sbt启动sbt控制台。准备就绪后，您可以发出运行命令，并为您的Spark Master位置提供一个参数。即运行本地 [ 5 ]

（同样，本文结尾处有一个截屏视频，显示了运行此命令的示例。请参见下面的参考资料。）

根据您的日志级别，您应该从SparkCassandra代码中看到各种SparkCassandra输出。 Cassandra的这些控制台输出指示成功。哦耶。现在跟我说。哦耶

在sbt控制台中运行代码是快速进行和测试更改的便捷方法。在我开发此代码时，在一个窗口中打开了一个终端，在另一个窗口中打开了一个编辑器。每当我更改并保存Scala源代码时，都可以在sbt控制台中重新运行。

因此，现在让我们说到要部署此Spark程序的地步。让我们在下一部分中查找。

第4部分：组装Spark Cassandra Scala代码并部署到Spark集群

要构建一个jar并将其部署到一个Spark集群，我们需要对build.sbt文件进行一些小的更改。从上面的代码中您可能已经注意到，文件中有注释，指示需要更改的内容。我们应该取消注释此行：

// "org.apache.spark" %% "spark-sql" % "1.6.1" % "provided",

并注释掉这一行：

"org.apache.spark" %% "spark-sql" % "1.6.1",

然后，我们可以从命令行运行sbt assembly以生成Spark可部署的jar。如果使用样本build.sbt文件，则将生成target / scala-2.10 / spark-cassandra-example-assembly-1.0.jar

要进行部署，请使用带有适当参数的spark-submit ；即

spark-submit --class "com.supergloo.SparkCassandra" --master spark://todd-mcgraths-macbook-pro.local:7077 ./target/scala-2.10/spark-cassandra-example-assembly-1.0.jarspark-submit --class "com.supergloo.SparkCassandra" --master spark://todd-mcgraths-macbook-pro.local:7077 ./target/scala-2.10/spark-cassandra-example-assembly-1.0.jar

结论

所以你怎么看？当您运行代码时，您可以看到最具攻击性的国王和遭受攻击最多的国王。如果不放弃，我认为可以争论曼斯·雷德（Mance Rayder）是否应与受攻击最严重的名单上的 Renly Baratheon并列。但是，这并不是本教程的重点。至于代码和设置，您对下一步有任何疑问，意见或建议吗？

Spark Cassandra教程截屏

在以下截屏视频中，我完成了本教程中描述的步骤。请继续关注，因为在截屏视频的末尾有blooper镜头。因为我的意思是，为什么不吹牛。

Spark Cassandra教程资源

所有源代码，包括我使用上面在Apache Spark Cassandra上描述的perl脚本清理过的wars.csv文件，示例代码
https://github.com/datastax/spark-cassandra-connector
带有Cassandra连接器的DataFrames
权力游戏数据： https : //github.com/chrisalbon/war_of_the_five_kings_dataset

翻译自: https://www.javacodegeeks.com/2016/05/apache-spark-cassandra-game-thrones.html