Spark SQL ガイド

Spark SQL アップグレード ガイド

Spark SQL 2.3から2.4へのアップグレード

Query Spark 2.3 以前の結果 Spark 2.4の結果 Remarks
SELECT
array_contains(array(1), 1.34D); 		true false Spark2.4では、左と右のパラメータがそれぞれ配列(double)とdouble型に昇格されます。
SELECT
array_contains(array(1), '1'); 		true integerの型は損失が少ないやり方でstring型に昇格できないため、AnalysisExceptionが投げられます。 ユーザは明示的なcastを使うことができます
SELECT
array_contains(array(1), 'anystring'); 		null integerの型は損失が少ないやり方でstring型に昇格できないため、AnalysisExceptionが投げられます。 ユーザは明示的なcastを使うことができます

Spark SQL 2.3.0から2.3.1以上へのアップグレード

Spark SQL 2.2 から 2.3 へのアップグレード

Spark SQL 2.1から2.2へのアップグレード

Spark SQL 2.0から2.1へのアップグレード

Spark SQL 1.6から2.0へのアップグレード

Spark SQL 1.5から1.6へのアップグレード

   ./sbin/start-thriftserver.sh \
     --conf spark.sql.hive.thriftServer.singleSession=true \
     ...

Spark SQL 1.4から1.5へのアップグレード

Spark SQL 1.3から1.4へのアップグレード

DataFrame データの reader/writer インタフェース

ユーザのフィードバックにより、(SQLContext.read)の中のデータを読み込み、(DataFrame.write)へ書き込むための新しくもっと柔軟なAPIを作成しました。古いAPIは非推奨にされました (例えば、SQLContext.parquetFile, SQLContext.jsonFile)。

SQLContext.read ( Scala, Java, Python ) および DataFrame.write ( Scala, Java, Python ) についての詳細な情報はAPIドキュメントを見てください。

グルーピング カラムを保持するDataFrame.groupBy

ユーザフィードバックに基づいて、DataFrameの結果のグルーピング カラムを保持するためにDataFrame.groupBy().agg() のデフォルトの挙動を変更しました。1.3での挙動を維持するためには、spark.sql.retainGroupColumnsを falseに設定してください。

// In 1.3.x, in order for the grouping column "department" to show up,
// it must be included explicitly as part of the agg function call.
df.groupBy("department").agg($"department", max("age"), sum("expense"))

// In 1.4+, grouping column "department" is included automatically.
df.groupBy("department").agg(max("age"), sum("expense"))

// Revert to 1.3 behavior (not retaining grouping column) by:
sqlContext.setConf("spark.sql.retainGroupColumns", "false")
// In 1.3.x, in order for the grouping column "department" to show up,
// it must be included explicitly as part of the agg function call.
df.groupBy("department").agg(col("department"), max("age"), sum("expense"));

// In 1.4+, grouping column "department" is included automatically.
df.groupBy("department").agg(max("age"), sum("expense"));

// Revert to 1.3 behavior (not retaining grouping column) by:
sqlContext.setConf("spark.sql.retainGroupColumns", "false");
import pyspark.sql.functions as func

# In 1.3.x, in order for the grouping column "department" to show up,
# it must be included explicitly as part of the agg function call.
df.groupBy("department").agg(df["department"], func.max("age"), func.sum("expense"))

# In 1.4+, grouping column "department" is included automatically.
df.groupBy("department").agg(func.max("age"), func.sum("expense"))

# Revert to 1.3.x behavior (not retaining grouping column) by:
sqlContext.setConf("spark.sql.retainGroupColumns", "false")

DataFrame.withColumn上での挙動の変更

1.4より前は、DataFrame.withColumn() はカラムの追加のみをサポートします。同じ名前の既存のカラムがあったとしても、結果のデータフレーム内の指定された名前を持つ新しいカラムとして、そのカラムが常に追加されるでしょう。1.4から、DataFrame.withColumn() は既存のすべてのカラムの名前と異なるカラム名の追加、あるいは同じ名前の既存のカラムの置きかえをサポートします。

この変更はScala APIだけのもので、PySparkあるいはSparkRのものでは無いことに注意してください。

Spark SQL 1.0-1.2 から 1.3 へのアップグレード

Spark 1.3. では、Spark SQLから "Alpha"のラベルが取り除かれ、この一環として利用可能なAPIの整理が行われました。Spark1.3以降は、Spark SQLは1.Xシリーズの他のリリースとバイナリ互換性を提供するでしょう。この互換性の保証には、明示的に安定していないと印を付けられている(つまり、DeveloperAPI あるいは Experimental)APIは含まれません。

SchemaRDDからデータフレームへの変更

Spark SQL 1.3にアップグレードした時にユーザが気づく最も大きな変更は、SchemaRDD が DataFrame に名前が変更されることです。これは、データフレームがもはやRDDから直接継承されないが、RDD自身の実装でRDDが提供するほとんどの機能を代わりに提供するため、重要です。データフレームは.rdd メソッドを呼ぶことでRDDに変換することも可能です。

In Scala, there is a type alias from SchemaRDD to DataFrame to provide source compatibility for some use cases. それでもDataFrame を代わりに使うためにコードを更新することをお勧めします。Java および Python ユーザはコードを更新する必要は無いでしょう。

JavaとScala APIの統一

Spark 1.3 以降には、Scala APIを模倣した別個のJava互換クラス (JavaSQLContext および JavaSchemaRDD) があります。Spark 1.3 ではJavaAPIおよびScala APIは統合されました。どちらかの言語のユーザは SQLContext および DataFrameを使用しなければなりません。In general these classes try to use types that are usable from both languages (i.e. Array instead of language-specific collections). 一般的なタイプが無い場合(例えば、クロージャーあるいはMapを渡す)、代わりに関数の上書きが使われます。

Additionally, the Java specific types API has been removed. ScalaおよびJavaのユーザはプログラム的にスキーマを説明するためにorg.apache.spark.sql.types にあるクラスを使わなければなりません。

dslパッケージの明示的な交換と削除の分離 (Scalaのみ)

import sqlContext._ から始まるSpark 1.3より前の多くのコード例、sqlContextからの全ての関数はスコープに入れられました。Spark 1.3では、RDDから DataFrameへの変換のための暗黙的な変換は、SQLContextの中に隔離しました。今はユーザは import sqlContext.implicits._を書かなければなりません。

Additionally, the implicit conversions now only augment RDDs that are composed of Products (i.e., case classes or tuples) with a method toDF, instead of applying automatically.

When using function inside of the DSL (now replaced with the DataFrame API) users used to import org.apache.spark.sql.catalyst.dsl. 代わりに公開データフレーム関数APIが使われるべきです: import org.apache.spark.sql.functions._.

DataTypeのための org.apache.spark.sql 内のタイプエイリアスの削除 (Scala のみ)

Spark 1.3 は、DataTypeのための基本sqlパッケージにあったタイプのエイリアスを削除しました。ユーザはorg.apache.spark.sql.typesにあるクラスを代わりにインポートしなければなりません。

UDF 登録はsqlContext.udfに移動しました (Java & Scala)

UDFを登録するために使われる関数、データフレーム DSLあるいはSQLの両方で使われる、は、SQLContext内のudfオブジェクトに移動されました。

sqlContext.udf.register("strLen", (s: String) => s.length())
sqlContext.udf().register("strLen", (String s) -> s.length(), DataTypes.IntegerType);

Python のUDF登録は変更されません。

Python DataType はもうシングルトンではありません

Pythonでデータタイプを使う場合は、シングルトンを参照する代わりにそれら(つまり StringType()) を構築する必要があるでしょう。

TOP
inserted by FC2 system