Spark のデータをPowerShell でMySQL にレプリケーションする方法

古川えりか
コンテンツスペシャリスト

PowerShell のシンプルなスクリプトで、Spark のデータをMySQL データベースにレプリケーション（複製）する方法を紹介します。

CData Cmdlets for SparkSQL を使えば、PowerShell からSpark のデータデータにリアルタイムで連携できます。データ同期などのタスクの連携にぴったりの製品です。本記事では、PowerShell からCData Cmdlets for SparkSQL およびCData Cmdlets for MySQL を使って、同期スクリプトを作成して実行します。

まずは、PowerShell でSpark への接続を行います。レプリケーションは4つのステップがあります。

SparkSQL への接続

SparkSQL への接続を確立するには以下を指定します。

Server：SparkSQL をホストするサーバーのホスト名またはIP アドレスに設定。
Port：SparkSQL インスタンスへの接続用のポートに設定。
TransportMode：SparkSQL サーバーとの通信に使用するトランスポートモード。有効な入力値は、BINARY およびHTTP です。デフォルトではBINARY が選択されます。
AuthScheme：使用される認証スキーム。有効な入力値はPLAIN、LDAP、NOSASL、およびKERBEROS です。デフォルトではPLAIN が選択されます。

Databricks への接続

Databricks クラスターに接続するには、以下の説明に従ってプロパティを設定します。Note：必要な値は、「クラスター」に移動して目的のクラスターを選択し、「Advanced Options」の下にある「JDBC/ODBC」タブを選択することで、Databricks インスタンスで見つけることができます。

Server：Databricks クラスターのサーバーのホスト名に設定。
Port：443
TransportMode：HTTP
HTTPPath：Databricks クラスターのHTTP パスに設定。
UseSSL：True
AuthScheme：PLAIN
User：'token' に設定。
Password：パーソナルアクセストークンに設定（値は、Databricks インスタンスの「ユーザー設定」ページに移動して「アクセストークン」タブを選択することで取得できます）。

Spark のデータの取得

モジュールのインストール:
```
Install-Module SparkSQLCmdlets
```

Spark への接続:

  $sparksql = Connect-SparkSQL  -Server $Server

取得ターゲットのリソースの取得:

  $data = Select-SparkSQL -Connection $sparksql -Table "Customers"

Invoke-SparkSQL cmdlet を使って、SQL-92 クエリを使用することもできます:

  $data = Invoke-SparkSQL -Connection $sparksql -Query 'SELECT * FROM Customers WHERE Country = @Country' -Params @{'@Country'='US'}

戻り値からカラム名のリストを保存します。

  $columns = ($data | Get-Member -MemberType NoteProperty | Select-Object -Property Name).Name

Spark のデータをMySQL データベースにレプリケーション

カラム名を指定できるようにして、データをMySQL データベースにレプリケーションします。

モジュールのインストール:
```
Install-Module MySQLCmdlets
```
MySQL DB に、MySQL Server 名、ユーザー、パスワード、レプリケーション先のデータベース名を指定して、接続します:
```
  $mysql = Connect-MySQL -User $User -Password $Password -Database $Database -Server $Server -Port $Port
  
```
Spark、保存された値、そしてAdd-MySQL Cmdlet を使って、MySQL にデータを1レコードずつ挿入します。この例では、MySQL 側のテーブルは、Spark のリソース（Customers）と同じテーブル名を持っている必要があります。
```
  $data | % {
    $row = $_
    $values = @()
    $columns | % {
      $col = $_
      $values += $row.$($col)
    }
    Add-MySQL -Connection $mysql -Table "Customers" -Columns $columns -Values $values
  }
  
```

次回以降のレプリケーションをシンプルに実現

一度PowerShell でSpark とMySQL に接続したら、次からは1行のコマンドでレプリケーションを実施できます:

  Select-SparkSQL -Connection $sparksql -Table "Customers" | % {
    $row = $_
    $values = @()
    $columns | % {
      $col = $_
      $values += $row.$($col)
    }
    Add-MySQL -Connection $mysql -Table "Customers" -Columns $columns -Values $values
  }

別のPowerShell モジュールで、Spark を別のデータベースに複製する場合、Select-SparkSQL cmdlet のデータから、カラム、接続およびテーブルを除外しておきましょう。これらのデータはデータ移動のときだけ必要となるためです。
```
  $columns = ($data | Get-Member -MemberType NoteProperty | Select-Object -Property Name).Name | ? {$_ -NotIn @('Columns','Connection','Table')}
  
```