Azure Sypanse Analyticsを一通り触ってみた

こんにちは、今日はAzure Synapse Analyticsについて学んだことをまとめておきたいと思います。

これまで、Microsoft Learnや資格取得を通して、Synapseの概要を学んできましたが、いかんせん触ったことがない状態では覚えるのにも限界がありますね・・・。

ということで、実際に触りながら覚えたいけど何かよいチュートリアルないかなぁと思って探していたら、ありました！

https://docs.microsoft.com/ja-jp/azure/synapse-analytics/get-started

今回はこちらを一通りなぞってみたので、画面キャプチャのメモも添えながら記録を残しておきたいと思います。

それではまいります。

Contents

Azure Synpase Analyticsって何？

そもそもSynpaseってなんだ？という方は以前こちらの記事をまとめていますので、参考になりましたら幸いです。

【Microsoft Azure】Azure Synapse Analyticsでできることを理解する

Synapseワークスペースの作成

最初はリソースとデータの作成です。以下のDocの手順に従って進めていきます。

参考URL：https://docs.microsoft.com/ja-jp/azure/synapse-analytics/get-started-create-workspace

事前にサブスクリプションでMicrosoft.Synapseリソースプロバイダを有効化しておきます。

Synpaseリソースの作成に進みます。

具体的には、ワークスペースリソースの作成になります。ここはお好きなものを指定いただけます。

続いてセキュリティの設定。こちらもいったんは規定値のまま進めます。

続けてネットワーク。こちらも規定値で。

以上でリソースの作成が完了します。（10分ほどでデプロイが完了した記憶です）

デプロイが完了したら、続けてデータの作成です。

リソースの概要ブレードからSynapse Studioを開きます。

以下URLから、読み込むデータファイルをダウンロードしてきます。

https://azuresynapsestorage.blob.core.windows.net/sampledata/NYCTaxiSmall/NYCTripSmall.parquet

Synpase Studioのデータ＞リンク済＞アップロードから、先ほどのファイルをアップロードします。

すると、アップロードファイルのURLが取得できるようになるので、控えておきます（次のステップで使います）

以上でステップ１は終了です。

サーバレスSQLプールを使用してデータを分析する

続けて、SQLプールを使って先ほどのデータにクエリをかけてみたいと思います。

参考URL：https://docs.microsoft.com/ja-jp/azure/synapse-analytics/get-started-analyze-sql-on-demand

開発＞SQLスクリプトから、新規SQLスクリプトを作成します。

ここで、以下のSQLを実行します。

SELECT
    TOP 100 *
FROM
    OPENROWSET(
        BULK 'https://<STORAGE ACCOUNT NAME>.dfs.core.windows.net/<FILE STORAGE NAME>/NYCTripSmall.parquet',
        FORMAT='PARQUET'
    ) AS [result]

先ほどのデータが取得できました！

続けて、データ探索用のデータベースを作成します。先ほどのSQLスクリプト上で以下コマンドを実行していきます。

## 探索用データベースの作成
CREATE DATABASE DataExplorationDB 
                COLLATE Latin1_General_100_BIN2_UTF8

## データベースの切り替え
USE DataExplorationDB

## DataExplorationDBからユーティリティオブジェクトの作成
CREATE EXTERNAL DATA SOURCE ContosoLake
WITH ( LOCATION = 'https://contosolake.dfs.core.windows.net')

## DataExplorationDBでのユーザのログイン作成
CREATE LOGIN data_explorer WITH PASSWORD = 'My Very Strong Password 1234!';

## データベースユーザの作成＋Admin権限の付与 
CREATE USER data_explorer FOR LOGIN data_explorer;
GO
GRANT ADMINISTER DATABASE BULK OPERATIONS TO data_explorer;
GO

## データの探索
SELECT
    TOP 100 *
FROM
    OPENROWSET(
            BULK '/<FILE STORAGE NAME>/NYCTripSmall.parquet',
            DATA_SOURCE = 'ContosoLake',
            FORMAT='PARQUET'
    ) AS [result]

最後のクエリまで実行すると、無事データの取得ができました。

最後に、「すべて発行」をして変更を確定します。

Data Explorerを使用して分析する

続けて、Data Explorer上で分析する方法についても試してみます。SynapseではData Explorerとの統合もできるのですね。

参考URL：https://docs.microsoft.com/ja-jp/azure/synapse-analytics/get-started-analyze-data-explorer

このチュートリアルは上で用意してきたデータベースを使わないようでしたが・・。

最初に、管理＞データエクスプローラーから、データエクスプローラープールを作成します。

リソースの作作完了には10分ほどかかりました。

続いてData Explorerデータベースの作成です。

データ＞新しいリソースの追加より、Data Explorerデータベースの作成を行います。

以下のように適当な名前でデータベースを作成しておきます。

次にこのデータベースにデータを蓄積して、分析を行っていきたいと思います。

開発＞新しいリソースの作成＞KQLスクリプトを開いて、以下コマンドを実行していきます。

## Data Explorerテーブルの作成
.create table StormEvents (StartTime: datetime, EndTime: datetime, EpisodeId: int, EventId: int, State: string, EventType: string, InjuriesDirect: int, InjuriesIndirect: int, DeathsDirect: int, DeathsIndirect: int, DamageProperty: int, DamageCrops: int, Source: string, BeginLocation: string, EndLocation: string, BeginLat: real, BeginLon: real, EndLat: real, EndLon: real, EpisodeNarrative: string, EventNarrative: string, StormSummary: dynamic)

## Data Explorerテーブルへのデータのインジェスト
.ingest into table StormEvents 'https://kustosamplefiles.blob.core.windows.net/samplefiles/StormEvents.csv?sv=2019-12-12&ss=b&srt=o&sp=r&se=2022-09-05T02:23:52Z&st=2020-09-04T18:23:52Z&spr=https&sig=VrOfQMT1gUrHltJ8uhjYcCequEcfhjyyMX%2FSc3xsCy4%3D' with (ignoreFirstRecord=true)

## Data Explorerテーブルの検索
StormEvents
| sort by StartTime desc
| take 10

最後まで実行すると、以下のようにData Exlorerテーブルからデータの取得ができるようになりました！

最後に、発行を行って変更を確定させます。

Apache Sparkを使用して分析を行う

お次はApache Sparkを用いたデータ分析を行うシナリオです。

参考URL：https://docs.microsoft.com/ja-jp/azure/synapse-analytics/get-started-analyze-spark

「管理」から新しいSparkプールを作成します。

プールが作成できたら、開発＞新しいノートブックを開いて、先ほど作成したクラスターに接続します。

このノートブック上で以下のコマンドを実行します。

%%pyspark
df = spark.read.load('abfss://<FILE STORAGE NAME>@<STORAGE ACCOUNT NAME>.dfs.core.windows.net/NYCTripSmall.parquet', format='parquet')
display(df.limit(10))

すると、Jupyter Notebookとかと同じ要領で対話式でコードが実行されることが確認できました。

手順に従って以下も実行してみます。先ほどはparquetファイルからデータを読みだしてきていましたが、以下のようにSparkデータベースに保存して、それに対して操作を行うこともできるのですね。

## データをSparkデータベースに読み込む
%%pyspark
spark.sql("CREATE DATABASE IF NOT EXISTS nyctaxi")
df.write.mode("overwrite").saveAsTable("nyctaxi.trip")

## Sparkデータベースのデータを操作する
%%pyspark
df = spark.sql("SELECT * FROM nyctaxi.trip") 
display(df)

専用SQLプールを使用してデータ分析を行う

続けて、専用SQLプール上でデータ分析を行うシナリオです。

参考URL：https://docs.microsoft.com/ja-jp/azure/synapse-analytics/get-started-analyze-sql-pool

だんだん操作に慣れてきました、専用SQLプールの機能を使うなら、「管理」からSQLプールの作成を行います。Synapseを作成したときに規定で作成されるプールはサーバレスタイプなので、専有プールが必要な場合は、こうやって明示的に作成する必要があるのですね。

専用プールを作成したら、以下のSQL文を実行します。

IF NOT EXISTS (SELECT * FROM sys.objects O JOIN sys.schemas S ON O.schema_id = S.schema_id WHERE O.NAME = 'NYCTaxiTripSmall' AND O.TYPE = 'U' AND S.NAME = 'dbo')
CREATE TABLE dbo.NYCTaxiTripSmall
    (
     [DateID] int,
     [MedallionID] int,
     [HackneyLicenseID] int,
     [PickupTimeID] int,
     [DropoffTimeID] int,
     [PickupGeographyID] int,
     [DropoffGeographyID] int,
     [PickupLatitude] float,
     [PickupLongitude] float,
     [PickupLatLong] nvarchar(4000),
     [DropoffLatitude] float,
     [DropoffLongitude] float,
     [DropoffLatLong] nvarchar(4000),
     [PassengerCount] int,
     [TripDurationSeconds] int,
     [TripDistanceMiles] float,
     [PaymentType] nvarchar(4000),
     [FareAmount] numeric(19,4),
     [SurchargeAmount] numeric(19,4),
     [TaxAmount] numeric(19,4),
     [TipAmount] numeric(19,4),
     [TollsAmount] numeric(19,4),
     [TotalAmount] numeric(19,4)
    )
WITH
    (
    DISTRIBUTION = ROUND_ROBIN,
     CLUSTERED COLUMNSTORE INDEX
     -- HEAP
    )
GO

COPY INTO dbo.NYCTaxiTripSmall
(DateID 1, MedallionID 2, HackneyLicenseID 3, PickupTimeID 4, DropoffTimeID 5,
PickupGeographyID 6, DropoffGeographyID 7, PickupLatitude 8, PickupLongitude 9, 
PickupLatLong 10, DropoffLatitude 11, DropoffLongitude 12, DropoffLatLong 13, 
PassengerCount 14, TripDurationSeconds 15, TripDistanceMiles 16, PaymentType 17, 
FareAmount 18, SurchargeAmount 19, TaxAmount 20, TipAmount 21, TollsAmount 22, 
TotalAmount 23)
FROM 'https://<STORAGE ACCOUNT NAME>.dfs.core.windows.net/<FILE STORAGE NAME>/NYCTripSmall.parquet'
WITH
(
    FILE_TYPE = 'PARQUET'
    ,MAXERRORS = 0
    ,IDENTITY_INSERT = 'OFF'
)

すると、データがdbo.NYCTaxiTripSmallというテーブルに読み込まれます。

次に、データ＞SQLデータベース＞SQLPOOL1＞テーブル＞dbo.NYCTaxiTripSmallを右クリックして、「上位100行を選択」としたSQLスクリプトを作成します。

作成されたSQLスクリプトを以下のコードで置き換えて実行します。

SELECT PassengerCount,
      SUM(TripDistanceMiles) as SumTripDistance,
      AVG(TripDistanceMiles) as AvgTripDistance
FROM  dbo.NYCTaxiTripSmall
WHERE TripDistanceMiles > 0 AND PassengerCount > 0
GROUP BY PassengerCount
ORDER BY PassengerCount;二課しか

すると、結果が表示されるのですが、結果の表示を「グラフ」に切り替えると、以下のような表示になりました。可視化が簡単に行えるのは便利ですね！

ストレージアカウント内のデータを分析する

次はストレージアカウントのデータを分析します。

参考URL：https://docs.microsoft.com/ja-jp/azure/synapse-analytics/get-started-analyze-storage

さきほどApache Sparkを使った分析で使ったNotebookを開いて、新しいセルに以下のコマンドを打ちます。これで、nyctaxi.passengercountstatsテーブルの内容が、csvとparquet方式でストレージアカウントに出力されます。

%%pyspark
df = spark.sql("SELECT * FROM nyctaxi.passengercountstats")
df = df.repartition(1) # This ensures we'll get a single file during write()
df.write.mode("overwrite").csv("/NYCTaxi/PassengerCountStats_csvformat")
df.write.mode("overwrite").parquet("/NYCTaxi/PassengerCountStats_parquetformat")

結果を確認してみます。

データ＞リンク済から、プライマリストレージの中身をみてみると、新しいフォルダとファイルが生成されていることが確認できました。

続けて、このストレージ内のファイルを直接読み込んでデータ分析を行うシナリオを再現してみます。

以下のように実行すればOKです。これでストレージのファイルからDataframeにデータを読み込んで分析を開始することができました。

%%pyspark
abspath = 'abfss://<FILE STORAGE NAME>@<STORAGE ACCOUNT NAME>.dfs.core.windows.net/NYCTaxi/PassengerCountStats_parquetformat/part-00000-1f251a58-d8ac-4972-9215-8d528d490690-c000.snappy.parquet'
df = spark.read.load(abspath, format='parquet')
display(df.limit(10))