如何在spark中讀寫cassandra數據 ---- 分布式計算框架spark學習之六


 

由於預處理的數據都存儲在cassandra里面,所以想要用spark進行數據分析的話,需要讀取cassandra數據,並把分析結果也一並存回到cassandra;因此需要研究一下spark如何讀寫cassandra。

話說這個單詞敲起來好累,說是spark,其實就是看你開發語言是否有對應的driver了。

因為cassandra是datastax主打的,所以該公司也提供了spark的對應的driver了,見這里

我就參考它的demo,使用scala語言來測試一把。

 

1.執行代碼

//CassandraTest.scala
import org.apache.spark.{Logging, SparkContext, SparkConf} import com.datastax.spark.connector.cql.CassandraConnector
object CassandraTestApp { def main(args: Array[String]) {
    #配置spark,cassandra的ip,這里都是本機 val SparkMasterHost
= "127.0.0.1" val CassandraHost = "127.0.0.1" // Tell Spark the address of one Cassandra node: val conf = new SparkConf(true) .set("spark.cassandra.connection.host", CassandraHost) .set("spark.cleaner.ttl", "3600") .setMaster("local[12]") .setAppName("CassandraTestApp") // Connect to the Spark cluster: lazy val sc = new SparkContext(conf)
    //預處理腳本,連接的時候就執行這些 CassandraConnector(conf).withSessionDo { session
=> session.execute("CREATE KEYSPACE IF NOT EXISTS test WITH REPLICATION = {'class': 'SimpleStrategy', 'replication_factor': 1 }") session.execute("CREATE TABLE IF NOT EXISTS test.key_value (key INT PRIMARY KEY, value VARCHAR)") session.execute("TRUNCATE test.key_value") session.execute("INSERT INTO test.key_value(key, value) VALUES (1, 'first row')") session.execute("INSERT INTO test.key_value(key, value) VALUES (2, 'second row')") session.execute("INSERT INTO test.key_value(key, value) VALUES (3, 'third row')") }     

//加載connector import com.datastax.spark.connector._
// Read table test.kv and print its contents: val rdd = sc.cassandraTable("test", "key_value").select("key", "value") rdd.collect().foreach(row => println(s"Existing Data: $row")) // Write two new rows to the test.kv table: val col = sc.parallelize(Seq((4, "fourth row"), (5, "fifth row"))) col.saveToCassandra("test", "key_value", SomeColumns("key", "value")) // Assert the two new rows were stored in test.kv table: assert(col.collect().length == 2) col.collect().foreach(row => println(s"New Data: $row")) println(s"Work completed, stopping the Spark context.") sc.stop() } }

 

2.目錄結構

由於構建工具是用sbt,所以目錄結構也必須遵循sbt規范,主要是build.sbt 和 src目錄, 其它目錄會自動生成。

qpzhang@qpzhangdeMac-mini:~/scala_code/CassandraTest $ll
total 8
drwxr-xr-x   6 qpzhang  staff  204 11 26 12:14 ./
drwxr-xr-x  10 qpzhang  staff  340 11 25 17:30 ../
-rw-r--r-- 1 qpzhang staff 460 11 26 10:11 build.sbt
drwxr-xr-x   3 qpzhang  staff  102 11 25 17:42 project/ drwxr-xr-x 3 qpzhang staff 102 11 25 17:32 src/
drwxr-xr-x   6 qpzhang  staff  204 11 25 17:55 target/ qpzhang@qpzhangdeMac-mini:~/scala_code/CassandraTest $tree src/
src/
└── main
    └── scala
        └── CassandraTest.scala

 

qpzhang@qpzhangdeMac-mini:~/scala_code/CassandraTest $cat build.sbt

name := "CassandraTest"

version := "1.0"

scalaVersion := "2.10.4"

libraryDependencies += "org.apache.spark" %% "spark-core" % "1.5.2" % "provided"
libraryDependencies += "com.datastax.spark" %% "spark-cassandra-connector" % "1.5.0-M2"

assemblyMergeStrategy in assembly := {
  case PathList(ps @ _*) if ps.last endsWith ".properties" => MergeStrategy.first
  case x =>
    val oldStrategy = (assemblyMergeStrategy in assembly).value
    oldStrategy(x)
}

這里需要注意的是,sbt安裝的是當時最新版本 0.13 , 並且安裝了 assembly插件(這里要吐槽一下sbt,下載一坨坨的jar包,最好有FQ代理,否則下載等待時間很長)。

qpzhang@qpzhangdeMac-mini:~/scala_code/CassandraTest $cat ~/.sbt/0.13/plugins/plugins.sbt
addSbtPlugin("com.typesafe.sbteclipse" % "sbteclipse-plugin" % "2.5.0")
addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "0.14.1")

 

3.sbt編譯打包

在build.sbt 目錄下,使用sbt命令啟動。

然后使用 compile 命令進行編譯,使用assembly進行打包。

在次期間,遇到了sbt-assembly-deduplicate-error的問題,參考這里

> compile
[success] Total time: 0 s, completed 2015-11-26 10:11:50
>> assembly
[info] Including from cache: slf4j-api-1.7.5.jar
[info] Including from cache: metrics-core-3.0.2.jar
[info] Including from cache: netty-codec-4.0.27.Final.jar
[info] Including from cache: netty-handler-4.0.27.Final.jar
[info] Including from cache: netty-common-4.0.27.Final.jar
[info] Including from cache: joda-time-2.3.jar
[info] Including from cache: netty-buffer-4.0.27.Final.jar
[info] Including from cache: commons-lang3-3.3.2.jar
[info] Including from cache: jsr166e-1.1.0.jar
[info] Including from cache: cassandra-clientutil-2.1.5.jar
[info] Including from cache: joda-convert-1.2.jar
[info] Including from cache: netty-transport-4.0.27.Final.jar
[info] Including from cache: guava-16.0.1.jar
[info] Including from cache: spark-cassandra-connector_2.10-1.5.0-M2.jar
[info] Including from cache: cassandra-driver-core-2.2.0-rc3.jar
[info] Including from cache: scala-reflect-2.10.5.jar
[info] Including from cache: scala-library-2.10.5.jar
[info] Checking every *.class/*.jar file's SHA-1.
[info] Merging files...
[warn] Merging 'META-INF/INDEX.LIST' with strategy 'discard'
[warn] Merging 'META-INF/MANIFEST.MF' with strategy 'discard'
[warn] Merging 'META-INF/io.netty.versions.properties' with strategy 'first'
[warn] Merging 'META-INF/maven/com.codahale.metrics/metrics-core/pom.xml' with strategy 'discard'
[warn] Merging 'META-INF/maven/com.datastax.cassandra/cassandra-driver-core/pom.xml' with strategy 'discard'
[warn] Merging 'META-INF/maven/com.google.guava/guava/pom.xml' with strategy 'discard'
[warn] Merging 'META-INF/maven/com.twitter/jsr166e/pom.xml' with strategy 'discard'
[warn] Merging 'META-INF/maven/io.netty/netty-buffer/pom.xml' with strategy 'discard'
[warn] Merging 'META-INF/maven/io.netty/netty-codec/pom.xml' with strategy 'discard'
[warn] Merging 'META-INF/maven/io.netty/netty-common/pom.xml' with strategy 'discard'
[warn] Merging 'META-INF/maven/io.netty/netty-handler/pom.xml' with strategy 'discard'
[warn] Merging 'META-INF/maven/io.netty/netty-transport/pom.xml' with strategy 'discard'
[warn] Merging 'META-INF/maven/joda-time/joda-time/pom.xml' with strategy 'discard'
[warn] Merging 'META-INF/maven/org.apache.commons/commons-lang3/pom.xml' with strategy 'discard'
[warn] Merging 'META-INF/maven/org.joda/joda-convert/pom.xml' with strategy 'discard'
[warn] Merging 'META-INF/maven/org.slf4j/slf4j-api/pom.xml' with strategy 'discard'
[warn] Strategy 'discard' was applied to 15 files
[warn] Strategy 'first' was applied to a file
[info] SHA-1: d2cb403e090e6a3ae36b08c860b258c79120fc90
[info] Packaging /Users/qpzhang/scala_code/CassandraTest/target/scala-2.10/CassandraTest-assembly-1.0.jar ...
[info] Done packaging.
[success] Total time: 19 s, completed 2015-11-26 10:12:22

 

4.提交到spark,執行結果

 

qpzhang@qpzhangdeMac-mini:~/project/spark-1.5.2-bin-hadoop2.6 $./bin/spark-submit --class "CassandraTestApp" --master local[4] ~/scala_code/CassandraTest/target/scala-2.10/CassandraTest-assembly-1.0.jar
//...........................
5/11/26 11:40:23 INFO TaskSetManager: Starting task 0.0 in stage 0.0 (TID 0, localhost, NODE_LOCAL, 26660 bytes)
15/11/26 11:40:23 INFO Executor: Running task 0.0 in stage 0.0 (TID 0)
15/11/26 11:40:23 INFO Executor: Fetching http://10.60.215.42:57683/jars/CassandraTest-assembly-1.0.jar with timestamp 1448509221160
15/11/26 11:40:23 INFO CassandraConnector: Disconnected from Cassandra cluster: Test Cluster
15/11/26 11:40:23 INFO Utils: Fetching http://10.60.215.42:57683/jars/CassandraTest-assembly-1.0.jar to /private/var/folders/2l/195zcc1n0sn2wjfjwf9hl9d80000gn/T/spark-4030cadf-8489-4540-976e-e98eedf50412/userFiles-63085bda-aa04-4906-9621-c1cedd98c163/fetchFileTemp7487594
894647111926.tmp
15/11/26 11:40:23 INFO Executor: Adding file:/private/var/folders/2l/195zcc1n0sn2wjfjwf9hl9d80000gn/T/spark-4030cadf-8489-4540-976e-e98eedf50412/userFiles-63085bda-aa04-4906-9621-c1cedd98c163/CassandraTest-assembly-1.0.jar to class loader
15/11/26 11:40:24 INFO Cluster: New Cassandra host localhost/127.0.0.1:9042 added
15/11/26 11:40:24 INFO CassandraConnector: Connected to Cassandra cluster: Test Cluster
15/11/26 11:40:25 INFO Executor: Finished task 0.0 in stage 0.0 (TID 0). 2676 bytes result sent to driver
15/11/26 11:40:25 INFO TaskSetManager: Finished task 0.0 in stage 0.0 (TID 0) in 2462 ms on localhost (1/1)
15/11/26 11:40:25 INFO TaskSchedulerImpl: Removed TaskSet 0.0, whose tasks have all completed, from pool
15/11/26 11:40:25 INFO DAGScheduler: ResultStage 0 (collect at CassandraTest.scala:32) finished in 2.481 s
15/11/26 11:40:25 INFO DAGScheduler: Job 0 finished: collect at CassandraTest.scala:32, took 2.940601 s
Existing Data: CassandraRow{key: 1, value: first row} Existing Data: CassandraRow{key: 2, value: second row} Existing Data: CassandraRow{key: 3, value: third row} //....................
5/11/26 11:40:27 INFO TaskSchedulerImpl: Removed TaskSet 3.0, whose tasks have all completed, from pool
15/11/26 11:40:27 INFO DAGScheduler: ResultStage 3 (collect at CassandraTest.scala:41) finished in 0.032 s
15/11/26 11:40:27 INFO DAGScheduler: Job 3 finished: collect at CassandraTest.scala:41, took 0.046502 s
New Data: (4,fourth row) New Data: (5,fifth row)
Work completed, stopping the Spark context.

cassandra中的數據

cqlsh:test> select * from key_value ;

 key | value
-----+------------
   5 |  fifth row
   1 |  first row
   2 | second row
   4 | fourth row
   3 |  third row

(5 rows)

到此位置,還算順利,除了assembly 重復文件的問題,都還算ok。

 


注意!

本站转载的文章为个人学习借鉴使用,本站对版权不负任何法律责任。如果侵犯了您的隐私权益,请联系我们删除。



 
粤ICP备14056181号  © 2014-2021 ITdaan.com