JDBC から他のデータベースへ

• データソースオプション
• データ型マッピング
  • MySQL からの Spark SQL データ型のマッピング
  • MySQL への Spark SQL データ型のマッピング
  • PostgreSQL からの Spark SQL データ型のマッピング
  • PostgreSQL への Spark SQL データ型のマッピング
  • Oracle からの Spark SQL データ型のマッピング
  • Oracle への Spark SQL データ型のマッピング
  • Microsoft SQL Server からの Spark SQL データ型のマッピング
  • Microsoft SQL Server への Spark SQL データ型のマッピング
  • DB2 からの Spark SQL データ型のマッピング
  • DB2 への Spark SQL データ型のマッピング
  • Teradata からの Spark SQL データ型のマッピング
  • Teradata への Spark SQL データ型のマッピング

Spark SQL には、JDBC を使用して他のデータベースからデータを読み取ることができるデータソースも含まれています。この機能は、JdbcRDD を使用するよりも優先されるべきです。これは、結果が DataFrame として返され、Spark SQL で簡単に処理したり、他のデータソースと結合したりできるためです。また、JDBC データソースは、ユーザーが ClassTag を提供する必要がないため、Java や Python からの使用も容易です。（これは、他のアプリケーションが Spark SQL を使用してクエリを実行できるようにする Spark SQL JDBC サーバーとは異なります。）

開始するには、Spark クラスパスに、使用するデータベースの JDBC ドライバーを含める必要があります。たとえば、Spark Shell から postgres に接続するには、次のコマンドを実行します。

./bin/spark-shell --driver-class-path postgresql-9.4.1207.jar --jars postgresql-9.4.1207.jar

データソースオプション

Spark は、JDBC の以下の大文字/小文字を区別しないオプションをサポートしています。JDBC のデータソースオプションは、以下のように設定できます。

• DataFrameReader の
  • DataFrameReader
  • DataFrameWriter
• CREATE TABLE USING DATA_SOURCE の OPTIONS 句

接続プロパティについては、データソースオプションで JDBC 接続プロパティを指定できます。user および password は、通常、データソースへのログインのための接続プロパティとして提供されます。

Kerberos 認証と keytab の組み合わせは、JDBC ドライバーで常にサポートされているわけではないことに注意してください。
keytab および principal 設定オプションを使用する前に、次の要件が満たされていることを確認してください。

• 含まれている JDBC ドライバーのバージョンが、keytab を使用した Kerberos 認証をサポートしている。
• 使用されているデータベースでサポートされている組み込み接続プロバイダーが存在する。

以下のデータベースには組み込み接続プロバイダーがあります。

• DB2
• MariaDB
• MS Sql
• Oracle
• PostgreSQL

要件が満たされていない場合は、カスタム認証を処理するために JdbcConnectionProvider 開発者 API を検討してください。

# Note: JDBC loading and saving can be achieved via either the load/save or jdbc methods
# Loading data from a JDBC source
jdbcDF = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:postgresql:dbserver") \
    .option("dbtable", "schema.tablename") \
    .option("user", "username") \
    .option("password", "password") \
    .load()

jdbcDF2 = spark.read \
    .jdbc("jdbc:postgresql:dbserver", "schema.tablename",
          properties={"user": "username", "password": "password"})

# Specifying dataframe column data types on read
jdbcDF3 = spark.read \
    .format("jdbc") \
    .option("url", "jdbc:postgresql:dbserver") \
    .option("dbtable", "schema.tablename") \
    .option("user", "username") \
    .option("password", "password") \
    .option("customSchema", "id DECIMAL(38, 0), name STRING") \
    .load()

# Saving data to a JDBC source
jdbcDF.write \
    .format("jdbc") \
    .option("url", "jdbc:postgresql:dbserver") \
    .option("dbtable", "schema.tablename") \
    .option("user", "username") \
    .option("password", "password") \
    .save()

jdbcDF2.write \
    .jdbc("jdbc:postgresql:dbserver", "schema.tablename",
          properties={"user": "username", "password": "password"})

# Specifying create table column data types on write
jdbcDF.write \
    .option("createTableColumnTypes", "name CHAR(64), comments VARCHAR(1024)") \
    .jdbc("jdbc:postgresql:dbserver", "schema.tablename",
          properties={"user": "username", "password": "password"})

// Note: JDBC loading and saving can be achieved via either the load/save or jdbc methods
// Loading data from a JDBC source
val jdbcDF = spark.read
  .format("jdbc")
  .option("url", "jdbc:postgresql:dbserver")
  .option("dbtable", "schema.tablename")
  .option("user", "username")
  .option("password", "password")
  .load()

val connectionProperties = new Properties()
connectionProperties.put("user", "username")
connectionProperties.put("password", "password")
val jdbcDF2 = spark.read
  .jdbc("jdbc:postgresql:dbserver", "schema.tablename", connectionProperties)
// Specifying the custom data types of the read schema
connectionProperties.put("customSchema", "id DECIMAL(38, 0), name STRING")
val jdbcDF3 = spark.read
  .jdbc("jdbc:postgresql:dbserver", "schema.tablename", connectionProperties)

// Saving data to a JDBC source
jdbcDF.write
  .format("jdbc")
  .option("url", "jdbc:postgresql:dbserver")
  .option("dbtable", "schema.tablename")
  .option("user", "username")
  .option("password", "password")
  .save()

jdbcDF2.write
  .jdbc("jdbc:postgresql:dbserver", "schema.tablename", connectionProperties)

// Specifying create table column data types on write
jdbcDF.write
  .option("createTableColumnTypes", "name CHAR(64), comments VARCHAR(1024)")
  .jdbc("jdbc:postgresql:dbserver", "schema.tablename", connectionProperties)

// Note: JDBC loading and saving can be achieved via either the load/save or jdbc methods
// Loading data from a JDBC source
Dataset<Row> jdbcDF = spark.read()
  .format("jdbc")
  .option("url", "jdbc:postgresql:dbserver")
  .option("dbtable", "schema.tablename")
  .option("user", "username")
  .option("password", "password")
  .load();

Properties connectionProperties = new Properties();
connectionProperties.put("user", "username");
connectionProperties.put("password", "password");
Dataset<Row> jdbcDF2 = spark.read()
  .jdbc("jdbc:postgresql:dbserver", "schema.tablename", connectionProperties);

// Saving data to a JDBC source
jdbcDF.write()
  .format("jdbc")
  .option("url", "jdbc:postgresql:dbserver")
  .option("dbtable", "schema.tablename")
  .option("user", "username")
  .option("password", "password")
  .save();

jdbcDF2.write()
  .jdbc("jdbc:postgresql:dbserver", "schema.tablename", connectionProperties);

// Specifying create table column data types on write
jdbcDF.write()
  .option("createTableColumnTypes", "name CHAR(64), comments VARCHAR(1024)")
  .jdbc("jdbc:postgresql:dbserver", "schema.tablename", connectionProperties);

# Loading data from a JDBC source
df <- read.jdbc("jdbc:postgresql:dbserver", "schema.tablename", user = "username", password = "password")

# Saving data to a JDBC source
write.jdbc(df, "jdbc:postgresql:dbserver", "schema.tablename", user = "username", password = "password")

CREATE TEMPORARY VIEW jdbcTable
USING org.apache.spark.sql.jdbc
OPTIONS (
  url "jdbc:postgresql:dbserver",
  dbtable "schema.tablename",
  user 'username',
  password 'password'
)

INSERT INTO TABLE jdbcTable
SELECT * FROM resultTable

データ型マッピング

MySQL からの Spark SQL データ型のマッピング

以下の表は、組み込みの jdbc データソースと MySQL Connector/J をアクティブな JDBC ドライバーとして使用して MySQL テーブルからデータを読み取る際に、MySQL データ型から Spark SQL データ型へのデータ型変換を示しています。Maria Connector/J など、MySQL に接続するために利用可能な異なる JDBC ドライバーでは、マッピングルールが異なる場合があることに注意してください。

MySQL への Spark SQL データ型のマッピング

以下の表は、組み込みの jdbc データソースと MySQL Connector/J をアクティブな JDBC ドライバーとして使用して MySQL テーブルにデータを作成、変更、または書き込む際に、Spark SQL データ型から MySQL データ型へのデータ型変換を示しています。

Maria Connector/J など、MySQL に接続するために利用可能な異なる JDBC ドライバーでは、マッピングルールが異なる場合があることに注意してください。

以下の Spark Catalyst データ型は、適切な MySQL 型ではサポートされていません。

• DayTimeIntervalType
• YearMonthIntervalType
• CalendarIntervalType
• ArrayType
• MapType
• StructType
• UserDefinedType
• NullType
• ObjectType
• VariantType

PostgreSQL からの Spark SQL データ型のマッピング

以下の表は、組み込みの jdbc データソースと PostgreSQL JDBC Driver をアクティブな JDBC ドライバーとして使用して PostgreSQL テーブルからデータを読み取る際に、PostgreSQL データ型から Spark SQL データ型へのデータ型変換を示しています。異なる JDBC ドライバーや異なるバージョンでは、わずかに異なる結果になる場合があることに注意してください。

PostgreSQL への Spark SQL データ型のマッピング

以下の表は、組み込みの jdbc データソースと PostgreSQL JDBC Driver をアクティブな JDBC ドライバーとして使用して PostgreSQL テーブルにデータを作成、変更、または書き込む際に、Spark SQL データ型から PostgreSQL データ型へのデータ型変換を示しています。

以下の Spark Catalyst データ型は、適切な PostgreSQL 型ではサポートされていません。

• DayTimeIntervalType
• YearMonthIntervalType
• CalendarIntervalType
• ArrayType - 要素型が上記にリストされていない場合。
• MapType
• StructType
• UserDefinedType
• NullType
• ObjectType
• VariantType

Oracle からの Spark SQL データ型のマッピング

以下の表は、組み込みの jdbc データソースと Oracle JDBC をアクティブな JDBC ドライバーとして使用して Oracle テーブルからデータを読み取る際に、Oracle データ型から Spark SQL データ型へのデータ型変換を示しています。

Oracle への Spark SQL データ型のマッピング

以下の表は、組み込みの jdbc データソースと Oracle JDBC をアクティブな JDBC ドライバーとして使用して Oracle テーブルにデータを作成、変更、または書き込む際に、Spark SQL データ型から Oracle データ型へのデータ型変換を示しています。

以下の Spark Catalyst データ型は、適切な Oracle 型ではサポートされていません。

• DayTimeIntervalType
• YearMonthIntervalType
• CalendarIntervalType
• ArrayType
• MapType
• StructType
• UserDefinedType
• NullType
• ObjectType
• VariantType

Microsoft SQL Server からの Spark SQL データ型のマッピング

以下の表は、組み込みの jdbc データソースと mssql-jdbc をアクティブな JDBC ドライバーとして使用して Microsoft SQL Server テーブルからデータを読み取る際に、Microsoft SQL Server データ型から Spark SQL データ型へのデータ型変換を示しています。

Microsoft SQL Server への Spark SQL データ型のマッピング

以下の表は、組み込みの jdbc データソースと mssql-jdbc をアクティブな JDBC ドライバーとして使用して Microsoft SQL Server テーブルにデータを作成、変更、または書き込む際に、Spark SQL データ型から Microsoft SQL Server データ型へのデータ型変換を示しています。

以下の Spark Catalyst データ型は、適切な SQL Server 型ではサポートされていません。

• DayTimeIntervalType
• YearMonthIntervalType
• CalendarIntervalType
• ArrayType
• MapType
• StructType
• UserDefinedType
• NullType
• ObjectType
• VariantType

DB2 からの Spark SQL データ型のマッピング

以下の表は、組み込みの jdbc データソースと IBM Data Server Driver For JDBC and SQLJ をアクティブな JDBC ドライバーとして使用して DB2 テーブルからデータを読み取る際に、DB2 データ型から Spark SQL データ型へのデータ型変換を示しています。

DB2 への Spark SQL データ型のマッピング

以下の表は、組み込みの jdbc データソースと IBM Data Server Driver For JDBC and SQLJ をアクティブな JDBC ドライバーとして使用して DB2 テーブルにデータを作成、変更、または書き込む際に、Spark SQL データ型から DB2 データ型へのデータ型変換を示しています。

以下の Spark Catalyst データ型は、適切な DB2 型ではサポートされていません。

• DayTimeIntervalType
• YearMonthIntervalType
• CalendarIntervalType
• ArrayType
• MapType
• StructType
• UserDefinedType
• NullType
• ObjectType
• VariantType

Teradata からの Spark SQL データ型のマッピング

以下の表は、組み込みの jdbc データソースと Teradata JDBC Driver をアクティブな JDBC ドライバーとして使用して Teradata テーブルからデータを読み取る際に、Teradata データ型から Spark SQL データ型へのデータ型変換を示しています。

Teradata への Spark SQL データ型のマッピング

以下の表は、組み込みの jdbc データソースと Teradata JDBC Driver をアクティブな JDBC ドライバーとして使用して Teradata テーブルにデータを作成、変更、または書き込む際に、Spark SQL データ型から Teradata データ型へのデータ型変換を示しています。

以下の Spark Catalyst データ型は、適切な Teradata 型ではサポートされていません。

• DayTimeIntervalType
• YearMonthIntervalType
• CalendarIntervalType
• ArrayType
• MapType
• StructType
• UserDefinedType
• NullType
• ObjectType
• VariantType