Sparkセッションの使用

Sparkセッション・オブジェクトは、Sparkアプリケーションのプライマリ・エントリー・ポイントで、データベース・テーブルでSQLクエリーを実行することができます。

Sparkセッションは、org.apache.spark.sql.SparkSessionのインスタンス内でカプセル化されます。セッション・オブジェクトには、Sparkマスター、Sparkアプリケーション、構成オプションに関する情報があります。

DSE Sparkシェルは、sparkという名前のSparkセッション・セッション・オブジェクトを自動的に構成して作成します。このオブジェクトを使用して、DataStax Enterpriseでデータベース・テーブルのクエリーを開始します。

spark.sql("SELECT * FROM keyspace.table_name")

注:

Spark 1.6以前では、個別のHiveContextとSQLContextオブジェクトがありました。Spark 2.0以降、SparkSessionは両方のオブジェクトをカプセル化します。

Sparkアプリケーションでは、他の基になるデータ・カタログを使用するために複数のセッションを使用できます。既存のSparkセッションを使用すると、newSessionメソッドを呼び出して、新しいセッションを作成できます。

val newSpark = spark.newSession()

Builder APIを使用したSparkセッションの構築

Builder APIを使用すると、Sparkセッションを手動で作成することができます。

import org.apache.spark.sql.SparkSession
val sparkSession = SparkSession.builder
  .master("dse://localhost?")
  .appName("my-spark-app")
  .enableHiveSupport()
  .config("spark.executor.logs.rolling.maxRetainedFiles", "3")
  .config("spark.executor.logs.rolling.strategy", "size")
  .config("spark.executor.logs.rolling.maxSize", "50000")
  .getOrCreate

Sparkセッションの停止

stopメソッドを使用して、Sparkセッションを終了します。

spark.stop

構成オプションの取得と設定

spark.conf.getおよびspark.conf.setメソッドを使用して、セッションのSpark構成オプションを取得または設定します。

spark.conf.set("spark.executor.logs.rolling.maxRetainedFiles", "3")
spark.conf.get("spark.executor.logs.rolling.maxSize")