言語

サポートされている言語とバージョン

Code Workbook は現在、Python、R、SQLの3つの言語をサポートしています。

現在サポートされているPythonのバージョンには、Python 3.6、Python 3.7、Python 3.8、Python 3.9が含まれます。Python 2はサポートされておらず、Python 2を使用する環境は解決に失敗します。Python 3.6はPython開発者ドキュメンテーションによるとend-of-lifeとされていますので、新しい利用可能なPythonのバージョンの使用を強く推奨します。

現在サポートされているRのバージョンには、R 3.5、R 3.6、R 4.0、R 4.1、R 4.2が含まれます。バージョンR 3.3とR 3.4はサポートされておらず、それぞれの環境は初期化に失敗します。

Code WorkbookでサポートされているSQLのバリアントはSpark SQLです。

Code Workbookのプロファイルで特定の言語を有効にする方法については、Code Workbookプロファイルの設定ドキュメンテーションのConda Environmentセクションを参照してください。サポートされている各言語の例は以下のPython、R、SQLへの導入で提供されます。

ワークブックで言語を有効にする

サポートされている言語を機能させるためには特定の設定が必要であり、以下のセクションで説明します。

Rの有効化

Code WorkbookでRトランスフォームを作成する能力を持つためには、次の2つの条件が満たされていなければなりません：

• R言語はユーザーのエンロールメントで特に有効にしなければなりません。有効化のためのアシスタンスはPalantirの代表者に連絡してください。
• パッケージvector-spark-module-rは現在ワークブックで使用されている環境に存在しなければなりません。これは以下のいずれかの方法で達成できます：
  • コントロールパネルのプロファイル設定でRチェックボックスを切り替える。これにより、vector-spark-module-rパッケージがプロファイルの環境に自動的に追加されます。
  • パッケージ追加ドロップダウンメニューを使用して、手動でvector-spark-module-rを環境に追加します。

詳細については、Code Workbookプロファイルの設定を参照してください。

Pythonの有効化

パッケージvector-spark-module-pyは、現在ワークブックで使用されている環境に存在しなければなりません。これは以下のいずれかの方法で達成できます：

• コントロールパネルのプロファイル設定でPythonチェックボックスを切り替える。これにより、vector-spark-module-pyパッケージが環境に自動的に追加されます。
• パッケージ追加ドロップダウンメニューを使用して、手動でvector-spark-module-pyを環境に追加します。

詳細については、Code Workbookプロファイルの設定を参照してください。

SQLの有効化

SQLトランスフォームは機能するために追加のパッケージを必要としません。その結果、SQLトランスフォームはデフォルトで任意のプロファイルで常に利用可能です。

Pythonへの導入

Pythonトランスフォーム

Pythonトランスフォームは、任意の数の入力、最大1つの出力、および任意に1つ以上のビジュアライゼーションを持つPython関数として定義されます。トランスフォームのエイリアスを関数引数として参照することで、Code Workbookは自動的にそのエイリアスの出力をトランスフォームの入力として渡します。Code Workbookでのトランスフォームについての詳細は、Transforms overviewドキュメンテーションを参照してください。

Pythonトランスフォームのシンプルな例としては、単一のPySpark DataFrameを入力として含め、PySparkの構文を使用してデータを変換し、変換されたSpark DataFrameを出力として持つことができます。

Copied!
1
2
3
4
5
def child(input_spark_dataframe):
    from pyspark.sql import functions as F
    # 'A'列が'value'と一致する行をフィルタリングします
    return input_spark_dataframe.filter(F.col('A') == 'value').withColumn('new_column', F.lit(2))
    # 新しい列 'new_column' を追加し、値を2に設定します

Spark と Pandas の Python での変換

Pythonトランスフォーム内では、Spark と Pandas のデータフレーム間の変換は簡単に行えます。

Copied!
1
2
3
4
# PySparkに変換
spark_df = spark.createDataFrame(pandas_df)
# pandasに変換
pandas_df = spark_df.toPandas()

以上のコメントはすでに日本語ですが、もし必要であれば詳細を追加します：

Copied!
1
2
3
4
# pandasのデータフレームをPySparkのデータフレームに変換
spark_df = spark.createDataFrame(pandas_df)
# PySparkのデータフレームをpandasのデータフレームに変換
pandas_df = spark_df.toPandas()

pandasに変換するということは、データをドライバに集めることを意味します。結果として、データのサイズはSparkモジュール上の利用可能なドライバメモリによって制約されます。大規模なデータセットを扱っている場合、まずSparkを使用してデータをフィルター処理し、集約し、それからpandasのDataFrameに集めることを検討するかもしれません。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
from pyspark.sql import functions as F
def filtering_before_pandas(input_spark_dataframe):
    # PySparkを使ってデータをフィルタリングします
    filtered_spark_df = input_spark_dataframe.select("name", "age").filter(F.col("age") <= 18)

    # pandasのデータフレームに変換し、ドライバにデータを収集します
    pandas_df = filtered_spark_df.toPandas()

    # pandasの操作を実行します
    mean_age = pandas_df["age"].mean()
    pandas_df["age_difference_to_mean"] = pandas_df["age"] - mean_age

    # 結果のデータフレームを出力します
    return pandas_df

Copied!
1
2
3
4
5
6
# pyspark.pandasをpという名前でインポートします
import pyspark.pandas as p 

# pandasのデータフレーム(df)をsparkのデータフレームに変換し、
# coalesce(1)を使ってパーティションを1つにまとめてから返します
return p.from_pandas(df).to_spark().coalesce(1)

Rへの入門

Rトランスフォーム

Rトランスフォームは、入力が任意の数、出力が最大で1つ、およびオプションで1つ以上の視覚化があるR関数として定義されます。トランスフォームのエイリアスを関数引数として参照すると、Code Workbookは自動的に、指定されたエイリアスの出力をトランスフォームの入力として渡します。Code Workbookでのトランスフォームについての詳細は、transforms overview のドキュメンテーションを参照してください。

Rトランスフォームの簡単な例としては、親となるR data.frameが1つあり、Rを使用してデータをトランスフォームし、出力としてR data.frameが1つあるものがあります。

Copied!
1
2
3
4
5
6
7
8
9
# 'child'という名前の関数を定義します。この関数は、引数としてRのデータフレームを受け取ります。
child <- def(r_dataframe) {
    # tidyverseパッケージを読み込みます。このパッケージには、データの操作と可視化に役立つ機能が含まれています。
    library(tidyverse)
    # 新しいデータフレームを作成します。このデータフレームは、元のデータフレームから列Aと列Bを選択し、列Aがtrueの行のみをフィルタリングし、新しい列（'new_column'）を追加して値を1000に設定します。
    new_df <- r_dataframe %>% dplyr::select(col_A, col_B) %>% dplyr::filter(col_A == true) %>% dplyr::mutate(new_column=1000)
    # 新しいデータフレームを返します。
    return(new_df)
}

Spark と R のデータフレーム間の変換

Rトランスフォーム内では、Spark DataFrameとRのdata.frame間の変換は直感的に行うことができます：

Copied!
1
2
3
4
5
# Spark DataFrameをRのdata.frameに変換
new_r_df <- SparkR::collect(spark_df)

# Rのdata.frameをSpark DataFrameに変換
 spark_df <- SparkR::as.DataFrame(r_df) 

上記のコードは既に自己説明的であり、日本語コメントを追加する必要はありません。ただし、それらを追加した場合は以下のようになります。

Copied!
1
2
3
4
5
# Spark DataFrameをRのdata.frameに変換します
new_r_df <- SparkR::collect(spark_df)

# Rのdata.frameをSpark DataFrameに変換します
 spark_df <- SparkR::as.DataFrame(r_df) 

R data.frame に変換することは、ドライバーへのデータの収集を意味します。その結果、データのサイズは Spark モジュール上の利用可能なドライバーメモリに制約されます。大量のデータセットを取り扱っている場合、最初に Spark を使用してデータをフィルター処理し、集約してから R data.frame に収集することをお勧めします。

Copied!
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
output_dataset <- function(spark_df) {
    library(tidyverse)

    # SparkR を使ってデータをフィルタリング
    input_dataset_filtered <- SparkR::select(spark_df, 'column_A', 'column_B')

    # R の data.frame に変換
    local_df <- SparkR::collect(input_dataset_filtered)

    # tidyverse の関数を使ってデータを変換
    local_df <- local_df %>% dplyr::filter(column_A == true) %>% dplyr::mutate(new_column = 1000)

    # R の data.frame を出力
    return(local_df) 
}

R トラブルシューティング

Code Workbook でインポートされたデータセットが R の data.frame として読み込まれる場合、データセットはドライバに集められることで Spark DataFrame から R の data.frame に変換されます。

• R の data.frame として任意の大きさのデータを読み込むことはできません。データのサイズは、Spark モジュール上のドライバメモリに制約されています。大きなデータを扱う場合は、まずデータセットを Spark DataFrame として読み込み、SparkR を使用してデータを小さく変換し、SparkR::collect() を呼び出して R の data.frame に変換することを検討してください。あるいは、R を使用する前に Python または SQL を使用してデータを小さく変換してください。
• R の data.frame に特定のデータ型を収集する際に、いくつかの既知の問題があります。収集時には、大半の場合、シリアライゼーションとデシリアライゼーションを高速化する r-arrow というライブラリを使用します。特に、r-arrow を使用する場合、Long、Array、Map、Struct、Datetime 型は変換できません。これらの行を削除するか、他のデータ型（String など）にキャストすることを検討してください。R の data.frame としてこれらの型を持つ入力を読み込もうとすると、インターフェースに警告が表示されます。

Code Workbook の R はシングルスレッドであり、同じ Spark モジュール上で一度に1つの R ジョブしか実行できません。複数の R ジョブを同時に開始すると、ジョブは直列に実行されます。キューに入っているジョブは "Code Workbook でキューイング中" と表示されます。

• 長時間実行されるジョブや、変換がデータセットとして保存されるジョブがある場合、バッチビルドを実行することをお勧めします。バッチビルドは独自の Spark モジュールで実行されるため、同じワークブックや Spark モジュールを共有する他のワークブックで継続的にイテレーションを行うことができます。

SQL 入門

Code Workbook でサポートされている SQL のバリアントは Spark SQL です。サポートされている入力および出力タイプは Spark DataFrame のみです。

SQL トランスフォームの簡単な例は、2つの入力 DataFrame を結合キーで結合することです。

Copied!
1
2
3
4
5
-- テーブルBからcol_Aとcol_Bを選択し、テーブルAから全ての列を選択します。
-- テーブルAとテーブルBをジョインします。ジョインの条件は、テーブルAのcol_CがテーブルBのcol_Cと等しいというものです。
SELECT table_b.col_A, table_b.col_B, table_a.*
FROM table_a
JOIN table_b ON table_a.col_C == table_B.col_C

SQLノードに親を追加するには、コード内でエイリアスを参照するだけでは不十分です。インプットバーを選択してUIを使用するか、**+**ボタンを使って子ノードを作成する必要があります。