同僚が最近、数十億レコードを顧客サイトのGeoEventの時空間ビッグデータストア（STBDS）に、ArcGIS Data Interoperabilityを使ってどのように移動するか尋ねてきました。これはユーティリティ会社のアーカイブされた車両位置情報で、Oracleジオデータベースに保存されています。毎日数百万の新しいイベントが入ってきています。アーカイブデータは70以上のテーブルに分かれています。<\/P>
STBDSのフィーチャサービスは通常のホストされたフィーチャサービスと同じREST APIを持つため、最初の計画は私の以前のブログで説明したパターンを活用し、すなわちアーカイブデータをポータルのシェイプファイルアイテムにETLし、その後ターゲットレイヤーのAppend RESTエンドポイントを使って非同期的にデータをロードするというものでした。さらにETLを複数の同時ジョブで並列化するという工夫もありました。この方法は、ポータルへ小さなトランザクション（デフォルトバッチサイズは1000フィーチャ）をストリーミングする際のネットワーク障害リスクを減らします。しかし顧客環境がこのワークフローをサポートするリリースではなかったため、結局ストリーミング方式を採用しましたが、この経験からネイティブに使える新しいクラウドウェアハウスやArcGIS Pro 2.9でビッグデータをETLする方法について考えるようになりました。<\/P>
基本的なメッセージは、大規模なデータセットをファイルとして複数の同時ジョブで移動し、スループットを最大化し輸送リスクを最小化してこれらのリフト＆シフト作業が可能だということです。さあ見てみましょう。<\/STRONG><\/P>Snowflake、Big Query、RedshiftなどのクラウドウェアハウスはArcGIS Pro 2.9でクエリや読み取りが可能ですが、標準ツールでは書き込みはできません。しかしArcGIS Data Interoperabilityはこれらのウェアハウスに空間データも含めて書き込み可能ですが、デフォルトモードはストリーミングであり、必要なスケールには合わないかもしれません。ここでは3つすべてのウェアハウスで使えるパターンをご紹介します:<\/P>Apache ParquetやCSVなどの形式で空間データを含む全データセットをETLするジオメトリは文字ベースの標準フォーマットでエンコードされる<\/LI><\/UL><\/LI>複数同時プロセスでETLを自動化する<\/STRONG><\/LI>ターゲット環境で必要なSQLやマクロコマンド以外はコードを書かない<\/STRONG><\/LI><\/UL>またノーコードブログスペースに潜むコーダー向けに少しだけ骨を投げます。以下をご覧ください😉。つまりParquetファイル作成に関するPythonのヒントです（注：Parquetファイルは2021年9月22日からArcGIS Onlineでサポートされているアイテムタイプです。ぜひ共有しましょう！）
このパターンでは数十億フィーチャが対象ですが、今回は説明用に控えめな230万ポイントフィーチャのみ使用しています。<\/P>
230万ポイントフィーチャ<\/span><\/span><\/P>私のデータは12個のフィーチャクラスに分かれており、任意の数にできます。このパターンは処理可能な別々の部分に分割されたデータで機能します。もしデータがモノリシックなら、自分で空間的に分割する（oriented fishnet anyone?)か、行位置によって埋められるバッチ識別子フィールドを追加して分割してください。そのフィールドは処理中に削除可能です。<\/P>Snowflake、Big Query、Redshiftウェアハウスについて言及しましたが、いずれの場合もターゲット環境が参照できる場所にParquetファイルをステージングし、そのParquetファイルからロードできます。空間データの場合、Parquetファイル内ジオメトリはターゲット環境が理解できる形式でエンコードされる必要があります（SnowflakeとBig QueryはGeoJSONとWKTをサポートし、RedshiftはWKTをサポート）。ここではGeoJSON形式でSnowflakeへロードする例のみ示します。私のデモデータはポイントジオメトリで、GeoJSON格納用フィールド幅は100です。ポリラインやポリゴンの場合は最もポイント数が多いフィーチャの幅を調査してください。例えば非常にポイント数が多いポリゴンを選択しGeoJSON化すると2,071,156文字になります:<\/P> <\/P>with arcpy.da.SearchCursor('NZ Property Titles','shape@') as cursor:
for row in cursor:
print(len(str(row[0].__geo_interface__)))
2071156 <\/P>Data InteroperabilityではGeoJSONの小数点精度制御が可能です。地理空間データには7桁程度が妥当で、その場合同じポリゴンは1,274,064文字になります。例えば最初の座標は (172.90677540000001,-41.12752416699993) から (172.9067754,-41.1275242) に変わります。すべてのバイト数が重要です！<\/P>注意: Big Query向けにはData InteroperabilityにGoogleBigQueryConnectorハブトランスフォーマーがありCSVからテーブルへのロードが可能です。これはParquet送信とbqコマンド環境によるロードより簡単かもしれませんが、このシナリオは調査していません。<\/P>私の特定ワークフローについて掘り下げましょう。秘訣は二つのETLツールを作成することです。一つ目はジョブ管理とWorkspaceRunnerトランスフォーマー呼び出し役で、二つ目ツール（実作業担当）を呼び出します。非常にシンプルです。こちらがLoadManager.fmwで引数としてジオデータベース内フィーチャクラス名一覧を受け取ります:<\/P>
LoadManager<\/span><\/span><\/P> WorkspaceRunnerは最大7つまでFMEプロセスを起動しジョブキュー消費までターゲットツール実行します。処理はCPUバウンドになる可能性が高く、ワーカープロセスがデータセットファイル抽出・エンコード・アップロードします。それぞれ2ジョブずつ実行させたため6プロセスで着実に処理しました。<\/P>
WorkspaceRunner<\/span><\/span><\/P> <\/P>こちらがLoadWorkerParquet.fmwです。<\/P>
LoadWorkerParquet<\/span><\/span>
非常にシンプルなツールで、ジオデータベース読み込み後ローカルParquetファイルを書き出し、その後Snowflakeへ送信してテーブルへコピーします。SQLExecutor部分をご自身で確認いただけますが理解補助として変数置換後Snowflake向けステートメント例はこちらです:
ParquetファイルがSnowflakeへ到達すると取り込み速度は非常に高速になります。ちなみに私は学びました ヘルプを読むことでこれを行うには、私はSnowflake DBAではありません。<\/P>どのようなパフォーマンスを期待すべきでしょうか？執筆時点では私も他の皆さんと同様に自宅に閉じこもっていますが、自宅のWiFiで6分間に230万の特徴量をSnowflakeにロードできるので、まともなコンピュータと有線ネットワークがあれば保守的に見積もって1時間あたり2500万のpoint特徴量は可能だと思います。もちろん、本番環境や非常に大きなジョブの場合は複数のコンピュータを使用でき、ターゲットとなるクラウドウェアハウスはスループットを処理するために確実にスケールします。<\/P>ブログのダウンロードには2番目のワーカーツール、LoadWorker.fmwも含めています。これは通常の方法で100K特徴量ごとにトランザクションを書き込む場合のパフォーマンス比較用で、はるかに遅かったです。<\/P>さて、core Pro 2.9に戻り、データはSnowflakeにロードされており、クエリを投げてスケールしたコンピュート体験を楽しめます。<\/P>
Catalogペイン内のSnowflake<\/span><\/P>parquetファイル作成用のPythonオプションについて言及しましたが、それはブログダウンロードにもありますがここにも示します：<\/P> <\/P># Pro 2.9+ の例：フィーチャクラスからparquetファイルを作成
# ジオメトリは'geom'フィールドにGeoJSONとしてエンコードされます
import arcpy
import pyarrow.parquet as pq
arcpy.env.overwriteOutput = True
# ソースフィーチャクラス
Canterbury = r"C:\Work\Parquet\Parquet.gdb\Canterbury"
# WGS84座標系でメモリ内フィーチャクラスを作成
with arcpy.EnvManager(outputCoordinateSystem='GEOGCS["GCS_WGS_1984",DATUM["D_WGS_1984",SPHEROID["WGS_1984",6378137.0,298.257223563]],PRIMEM["Greenwich",0.0],UNIT["Degree",0.0174532925199433]]',
geographicTransformations="NZGD_2000_To_WGS_1984_1"):
arcpy.conversion.ExportFeatures(
in_features=Canterbury,
out_features=r"memory\Canterbury")
# geomフィールドを追加（ポイント以外のジオメトリはより広いフィールドが必要です）
Canterbury = arcpy.management.AddField("memory\Canterbury","geom","TEXT",None,None,100,'',"NULLABLE","NON_REQUIRED",'').getOutput(0)
# GeoJSONを導出
with arcpy.da.UpdateCursor(Canterbury,['shape@','geom']) as cursor:
for row in cursor:
row[1] = str(row[0].__geo_interface__)
cursor.updateRow(row)
# テーブルを作成してジオメトリを削除
esriTable = arcpy.conversion.TableToTable(Canterbury,"memory","CanterburyTable").getOutput(0)
# arrowテーブルを作成
arrowTable = arcpy.da.TableToArrowTable(esriTable)
# parquetを書き込み
pq.write_table(arrowTable,r'C:\Work\Parquet\TitlesCanterbury.parquet',
version='1.0',
compression='SNAPPY')
<\/P>大規模なデータ移動を楽しんでください！<\/P> <\/P>
