CData Virtuality に対するデータ操作を Databricks で実施する方法

## 概要

Databricks (Spark) から jdbc にて CData Virtuality に接続して下記のデータ操作方法を説明します。3 つの処理パターンを 2 つの JDBC コネクター（CData Virtuality JDBC Driver と PostgreSQL JDBC Driver）を利用して実施する方法を紹介します。

1. 処理パターン
   1. CData Virtuality からデータ取得
   1. CData Virtuality へのデータ書き込み
   1. SQL 実行
1. 利用する JDBC コネクター
   1. CData Virtuality JDBC Driver
   1. PostgreSQL JDBC Driver

CData Virtuality は PostgreSQL プロトコルをサポートしています。2024年11月28日時点での検証では、データの読み込み処理は正常に動作しましたが、書き込み処理についてはエラーが発生することを確認しています。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/1f4369d5-33df-35f0-84e7-92e3fc68a50c.png)

> It uses the PostgreSQL protocol and has been specially designed for heavily multi-threaded applications.

引用元：[Python](https://docs.datavirtuality.com/v3/python#Python-AccessDataVirtualityServerfromPythonUsingpsycopg2)

> PostgreSQL プロトコルを使用し、マルチスレッド処理を重視したアプリケーション向けに特別に設計されています。

上記の翻訳

## JDBC Driver に関する情報

### 認証に利用するユーザーアカウントについて

CData Virtuality のアカウントがメールアドレスの場合には、`@`や`.`が`_`に変更する必要があります。`manabian@test.com`というメールアドレスの場合には、`manabian_test_com`のようなユーザーアカウントで認証を実施します。CData Virtuality の Preferences にて確認できます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/01e0849c-48d6-3e7d-1a5d-ff7bb6e00821.png)

### CData Virtuality JDBC Driver について

接続方法については、 CData 社のブログにて詳細が記載されています。本記事では SaaS版での検証を行っているため後者のパターンを実施しています。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/97eb8500-f45b-09b1-77d7-a5aae845a4b6.png)

引用元：[CData Virtuality - JDBCでの接続](https://www.cdata.com/jp/blog/cdv-jdbc)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/207b89b4-242a-0f74-fc62-1b5adb571997.png)

引用元：[CData Virtuality - JDBCでの接続](https://www.cdata.com/jp/blog/cdv-jdbc)

jdbc に渡すパラメータについては下記のドキュメントに記載されています。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/16a8a24f-1750-6214-3182-ed42b3018eab.png)

引用元：[JDBC Driver Installation and Connection](https://www.cdata.com/jp/virtuality/docs/v4/jdbc-driver-installation-and-connection.html)


Spark Dataframe にて jdbc によるデータ操作を行う際のパラメータについて下記のドキュメントに記載されております。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/1989415e-d04b-553c-dd58-f68a868902a7.png)

引用元：[JDBC To Other Databases - Spark 3.5.1 Documentation](https://spark.apache.org/docs/3.5.1/sql-data-sources-jdbc.html)


## 事前準備

### 1. CData Virtuality JDBC Driver を取得

下記のサイトから JDBC ドライバーをダウンロードします。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/ee0a3174-e6e9-4647-fa78-e9b78e056799.png)

引用元：[CData Virtuality | ODBC & JDBC Drivers](https://www.cdata.com/virtuality/clients/)

### 2. Snowflake にてテスト用のテーブルを作成

```sql
create or replace TABLE CDATA_01.SCHEMA_1.CDATA_TABLE_01 (
	TIMESTAMP_COL TIMESTAMP_NTZ(9)
);
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/8fef7b75-7ac2-a60e-086b-5f28ea516b7e.png)

### 3. CData Virtuality にて Snowflake を Sources に登録

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/3cde1006-6355-5243-2652-5057d01801bb.png)

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/fcebe614-4ebe-fafe-5bdf-cfa4337ef3bb.png)

### 4. CData Virtuality の jdbc を Databricks のクラスターにインストール

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/80c797c9-0924-5591-9b66-262d8a3741d7.png)

## データ操作方法

### 1-1. CData Virtuality JDBC Driver により CData Virtuality からデータ取得

```python
# CData Virtualtiy への接続情報を指定
driver = "com.datavirtuality.dv.jdbc.Driver"
database_host = "aws-ap-east-1.platform.datavirtuality.com"
jdbc_protocol = "mms"
database_port = "45006"
database_name = "datavirtuality"
url = f"jdbc:datavirtuality:{database_name}@{jdbc_protocol}://{database_host}:{database_port}"

# ユーザー名とパスワードを指定
user = "manabina_test_co_jp"
password = "2!zzzzzz"

# データ取得元テーブル名の指定
table = "snowflake_01.CDATA_01.SCHEMA_1.CDATA_TABLE_01"
```

```python
dv_df = (
    spark.read.format("jdbc")
    .option("driver", driver)
    .option("url", url)
    .option("dbtable", table)
    .option("user", user)
    .option("password", password)
    .load()
)
dv_df.limit(10).display()
```

### 1-2. CData Virtuality JDBC Driver により CData Virtuality へのデータ書き込み

append モードで書き込むことが可能です。

```python
from pyspark.sql.functions import current_timestamp

src_df = spark.createDataFrame([(1,)], ["id"]).withColumn("TIMESTAMP_COL", current_timestamp())
src_df = src_df.drop("id")

src_df.display()
```

```python
# CData Virtualtiy への接続情報を指定
driver = "com.datavirtuality.dv.jdbc.Driver"
database_host = "aws-ap-east-1.platform.datavirtuality.com"
jdbc_protocol = "mms"
database_port = "45006"
database_name = "datavirtuality"
url = f"jdbc:datavirtuality:{database_name}@{jdbc_protocol}://{database_host}:{database_port}"

# ユーザー名とパスワードを指定
user = "manabina_test_co_jp"
password = "2!zzzzzz"

# データ取得元テーブル名の指定
table = "snowflake_01.CDATA_01.SCHEMA_1.CDATA_TABLE_01"
```

```python
(
    src_df.write.format("jdbc")
    .mode("append")
    .option("truncate", "true")
    .option("driver", driver)
    .option("url", url)
    .option("dbtable", table)
    .option("user", user)
    .option("password", password)
    .save()
)
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/1d65af6a-5265-f3d1-8121-b12a635e6027.png)

overwrite モードで書き込むことは実施できないため、下記記事で紹介している方法を利用してデータを書き込むことができます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/b0aac38a-77da-c9cd-652f-27f44a5d86e3.png)

引用元：[CData Virtuality に対して Spark DataFrame にて overwrite により書き込む際のエラーへの対応方法 #Python - Qiita](https://qiita.com/manabian/items/04962aa5f8c737b35735)

### 1-3. CData Virtuality JDBC Driver により SQL 実行

```scala
%scala
import java.sql.DriverManager
import java.sql.Connection

var connection: Connection = null

try {
  val executed_sql = "INSERT INTO snowflake_01.CDATA_01.SCHEMA_1.CDATA_TABLE_01 SELECT NOW();;"

  val database_host = "aws-ap-east-1.platform.datavirtuality.com"
  val jdbc_protocol = "mms"
  val database_port = "45006"
  val database_name = "datavirtuality"
  val jdbcUsername = "manabina_test_co_jp"
  val jdbcPassword = "2!zzzzzz"
  val jdbcUrl = s"jdbc:datavirtuality:${database_name}@${jdbc_protocol}://${database_host}:${database_port};"

  connection = DriverManager.getConnection(jdbcUrl, jdbcUsername, jdbcPassword)
  val statement = connection.createStatement()
  statement.execute(executed_sql)

} finally {
  if (connection != null) {
    connection.close()
  }
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/cec57e99-a227-9436-0d00-12bda6878963.png)

### 2-1. PostgreSQL JDBC Driver により CData Virtuality からデータ取得

```python
# CData Virtualtiy への接続情報を指定
driver = "com.datavirtuality.dv.jdbc.Driver"
database_host = "aws-ap-east-1.platform.datavirtuality.com"
database_port = "45007"
database_name = "datavirtuality"

# ユーザー名とパスワードを指定
user = "manabina_test_co_jp"
password = "2!zzzzzz"

# データ取得元テーブル名の指定
table = "snowflake_01.CDATA_01.SCHEMA_1.CDATA_TABLE_01"
```

```python
dv_df = (
    spark.read.format("postgresql")
    .option("dbtable", table)
    .option("host", database_host)
    .option("port", database_port)
    .option("database", database_name)
    .option("user", user)
    .option("password", password)
    .load()
)

dv_df.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/fd82c0cf-0697-4ca0-a6d8-4fae85ee9353.png)

### 2-2. PostgreSQL JDBC Driver により CData Virtuality へのデータ書き込み

2024年11月28日時点ではエラーとなってしまいました。

### 2-3. PostgreSQL JDBC Driver により SQL 実行

```scala
%scala
import java.sql.DriverManager
import java.sql.Connection

var connection: Connection = null

try {
  val executed_sql = "INSERT INTO snowflake_01.CDATA_01.SCHEMA_1.CDATA_TABLE_01 SELECT NOW();;"

  val database_host = "aws-ap-northeast-1.platform.datavirtuality.com"
  val database_port = "45007"
  val database_name = "datavirtuality"
  val jdbcUsername = "manabina_test_co_jp"
  val jdbcPassword = "2!zzzzzz"
  val jdbcUrl = s"jdbc:postgresql://${database_host}:${database_port}/${database_name}"
  connection = DriverManager.getConnection(jdbcUrl, jdbcUsername, jdbcPassword)
  val statement = connection.createStatement()
  statement.execute(executed_sql)

} finally {
  if (connection != null) {
    connection.close()
  }
}
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/3a0ea773-b5ec-c9ff-adea-a58c845c1a19.png)

## 補足

### 1. PostgreSQL JDBC Driver により Spark Dataframe から書き込み処理を実行するとエラーが発生


```python
from pyspark.sql.functions import current_timestamp

src_df = spark.createDataFrame([(1,)], ["id"]).withColumn("TIMESTAMP_COL", current_timestamp())
src_df = src_df.drop("id")

src_df.display()
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/5ef7e1f4-4cc9-f8e9-5a74-923b1e302167.png)


```python
# CData Virtualtiy への接続情報を指定
driver = "com.datavirtuality.dv.jdbc.Driver"
database_host = "aws-ap-east-1.platform.datavirtuality.com"
database_port = "45007"
database_name = "datavirtuality"

# ユーザー名とパスワードを指定
user = "manabina_test_co_jp"
password = "2!zzzzzz"

# データ取得元テーブル名の指定
table = "snowflake_01.CDATA_01.SCHEMA_1.CDATA_TABLE_01"
```

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/be9e71ca-2a4e-fa87-85c1-80ec5c93a307.png)

```python
(
    src_df.write.format("postgresql")
    .mode("append")
    .option("dbtable", table)
    .option("host", database_host)
    .option("port", database_port)
    .option("database", database_name)
    .option("user", user)
    .option("password", password)
    .save()
)
```

> Py4JJavaError: An error occurred while calling o531.save.
: org.apache.spark.SparkException: Job aborted due to stage failure: Task 3 in stage 36.0 failed 4 times, most recent failure: Lost task 3.3 in stage 36.0 (TID 97) (10.139.64.10 executor 0): org.postgresql.util.PSQLException: ERROR: TEIID31100 Parsing error: Encountered "[*]SET[*] SESSION CHARACTERISTICS" at line 1, column 1.

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/d608b326-df8e-8780-270d-039b59c9891f.png)

```python
(
    src_df.write.format("postgresql")
    .mode("overwrite")
    .option("truncate", "true")
    .option("dbtable", table)
    .option("host", database_host)
    .option("port", database_port)
    .option("database", database_name)
    .option("user", user)
    .option("password", password)
    .save()
)
```

> Py4JJavaError: An error occurred while calling o546.save.
: org.postgresql.util.PSQLException: ERROR: TEIID31100 Parsing error: Encountered "[*]TRUNCATE[*] TABLE ONLY" at line 1, column 1.
Was expecting: <EOF> | "begin" | "{"

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/24031/6b6b449a-73aa-5b5b-b998-5b48e6021296.png)