【Spark】Spark-Shellの使用＋サンプル

現在Spark勉強中です。spark-shellの使い方と、Sparkの簡単なサンプルプログラムをメモとして残しておきます。

Spark-Shellの実行例

コマンド実行サンプル

$ sudo $SPARK_HOME/bin/spark-shell --packages datastax:spark-cassandra-connector:2.0.0-M2-s_2.11

confの説明

spark.cassandra.connection.host：cassandraのシードノード

spark.scheduler.allocation.file：Sparkの実行優先順位についてのファイルを指定

spark.cores.max：Sparkで使用するcoreの数

spark.executor.memory：executorプロセスあたりに使用するメモリ量

SparkContext：sparkのmasterを指定

サンプルソース

テーブルの情報を取得し、テーブル全件の件数を出力するサンプルです。

#SparkSQL
sc.stop

import org.apache.spark
import org.apache.spark._
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.SparkConf
import org.apache.spark.sql.SQLContext
import org.apache.spark.sql.cassandra
import org.apache.spark.sql.cassandra._
import com.datastax.spark
import com.datastax.spark._
import com.datastax.spark.connector
import com.datastax.spark.connector._
import com.datastax.spark.connector.cql
import com.datastax.spark.connector.cql._
import com.datastax.spark.connector.cql.CassandraConnector
import com.datastax.spark.connector.cql.CassandraConnector._

val conf = new SparkConf(true).set("spark.cassandra.connection.host", "10.222.**.***").set("spark.cassandra.auth.username", "otoraku").set("spark.cassandra.auth.password", "otorakupass").set("spark.scheduler.allocation.file", "/usr/local/bin/spark/conf/fairscheduler.xml.template").set("spark.cores.max","20").set("spark.executor.memory", "1g")
val sc = new SparkContext("spark://10.222.**.***:7077", "test1", conf)
sc.setLocalProperty("spark.scheduler.pool", "production")

import org.apache.spark.sql.SparkSession
val spark = SparkSession.builder().appName("Spark SQL basic example").config("spark.some.config.option", "some-value").getOrCreate()

import spark.implicits._
val sqlContext = new org.apache.spark.sql.SQLContext(sc)

val df = sqlContext.read.format("org.apache.spark.sql.cassandra").options(Map( "table" -> "c_streams_ymd", "keyspace" -> "otoraku" )).load()
df.createOrReplaceTempView("test")

val sqlDF = spark.sql("SELECT count(*) FROM test")

sqlDF.show()