SQL Server 使いのための Azure Synapse Analytics - Spark 入門

畠山大有 | Daiyu Hatakeyama | @dahatake Architect && Software Engineer && Applied Data Scientist Microsoft Japan SQL Server 使いのための Azure Synapse Analytics Spark 入門

HDFS 分散ファイルシステム“HDFS” (Hadoop Distributed File System)  複数のコンピューターを束ねて、一つの仮想的なファイルシステムを提供  Hadoop クラスタに属する全てのノードが同じ名前空間を共有  各ファイルは 64 MB ~ 256 MB 程度のブロックに分割されて、複数ノードに分散  ブロックサイズが大きいのは、シーケンシャルアクセスに特化しているため  各断片は最低 3 ノードに複製され、ノード障害への耐性を持つ  一つのファイルの複数の部分を、複数のコンピューターで同時に処理することで、処理を高速化することができる断片 1 断片 2 断片 3 断片 4 断片 1 断片 2 断片 3 断片 4 ファイル 1 断片 1 断片 2 断片 3 断片 4 ファイル 4 断片 1 断片 2 断片 3 断片 4

分散プログラミングモデル “MapReduce” 2013-08-26 23:59:03 W3SVC1 192.168.0.1 GET /… 2013-08-26 23:59:05 W3SVC1 192.168.0.1 GET /… 2013-08-26 23:59:25 W3SVC1 192.168.0.1 GET /… 2013-08-27 00:01:02 W3SVC1 192.168.0.1 GET /… 2013-08-27 00:02:04 W3SVC1 192.168.0.1 GET /…  名前は一見難しげですが、実際には単純です（だからこそスケーラビリティがあります）  例えば右のような Web のアクセスログがあるとして、  伝統的なテキスト処理ツールによる集計はこのようなものですが、 $ grep /Login.aspx access_log | sort | uniq –c > output.txt 対象ファイルの全行を頭から舐めて何らかの処理をする（この場合は検索）中間結果を集計のためにソートするソートされたデータを数え上げて結果を得る MapReduce はこういった処理をクラスター上で並列分散実行する仕組みです Map Reduce

SQL Query Data Frame Unresolved Logical Plan Logical Plan Optimized Logical Plan Selected Physical Plan RDDsPhysical Plans CostModel Catalog Analysis Logical Optimization Physical Planning Code Generation Catalyst 分散処理オプティマイザーインメモリ

項目 2013 年の記録 Hadoop Spark 100 TB (2014年) Spark 1 PB (2014年) データサイズ 102.5TB 100 TB 1 PB (≒1000 TB) 処理時間 72 分 23 分 234 分ノード数 2100 206 190 コア数 50,400 6,592 6,080

0 20 40 60 80 100 120 1 2 Series1

9 Standalone Scheduler YARN Mesos Spark SQL Spark Streaming MLlib GraphX PythonSQLR Scala Java

• 外部のデータベース、構造化ファイル、 Hive テーブル等、様々なデータソースにクエリを実行することができる • SQL と HiveQL の両方が使える • Python, Scala and Java からも実行できる • Spark structured streaming もサポートをオプティマイザとして採用

Spark での ETL Query の例 Read Table 1 Read Table 2 Join Filter Write

• カラムストア圧縮スキーマ Col. A Col. B a1 ... ak ak+1 ... an b1 ... bk bk+1 ... bn a1 ... ak b1 ... bk ak+1 ... an bk+1 ... bn Row group 1 Row group 2 Column chunk A1 Column chunk B1 Column chunk A2 Column chunk B2 Col. chunk A1 Col. chunk B1 Col. chunk A2 Col. chunk B2 Footer Header Data Header Data Page ... Row group 1 Col. chunk A1: 0, 1000 Col. chunk B1: 1000, 500 Row group 2 Col. chunk A2: 1500, 2000 Col. chunk B2: 3500, 500 ... 論理構造物理構造

Metric Result Comment Serialized size Parquet is 20-40% smaller than Bond Columnar format -> better compression Serialization throughput Parquet is 3-6x faster than Bond Cheaper compression codec Deserialization throughput Parquet is 1.5-3x faster than Bond Cheaper string encoding (UTF-8 vs UTF-16) Query throughput: point lookup Parquet is 4x faster than Bond Most columns are not deserialized Query throughput: Scrubbing Parquet is 4x faster than Bond Most columns are not deserialized Query throughput: LogsToMetrics Parquet is 1.5x faster than Bond Better deserialization throughput Query throughput: SecurityWorkflow Parquet is as fast as Bond Data format is not a bottleneck

Group By が Avro より 2.6倍程度高速全件スキャンで、Avro より 2倍以上高速ファイルサイズが Avro より 25% 小さい。圧縮していないCSVよりも 82%小さい

delta dataframe.write .format("parquet") .partitionBy(“date") .save("/path/to/parquet") CREATE TABLE events ( date DATE, eventId STRING, eventType STRING) USING parquet OPTIONS (path = ‘/path/to/parquet' ) PARTITIONED BY (date) CREATE TABLE events_delta USING delta LOCATION ‘/path/to/delta' dataframe.write .format(“delta") .partitionBy(“date") .save("/path/to/delta")

• Apache Spark 2.4 • Linux Foundation Delta Lake 0.4 support • .Net Core 3.0 support • Python 3.6 + Anaconda support • 他の Azure Synapse services との連携 • セキュリティとログイン • メタデータ • プロビジョニング • nteract ベースの notebooks • 高速な SQL pools へのデータロード • よくある利用シナリオ • データ準備 / Data Engineering / ETL • Spark ML / Azure ML integration • 効果的なリソースの使用 • 迅速に開始 • オートスケール。最小3 node • 自動一時停止 • 様々なプログラミング言語のサポート • .Net (C#), PySpark, Scala, Spark SQL, Java

SQL Server Synapse Spark Pool ファイル形式独自バイナリー : mdf,ndf,ldf Parquet などファイルの場所ファイルとして指定 HDFS: 実際には Azure Data Lake Storage Gen 2 スケールアップコンピューターのスペックアップコンピューターのスペックアップスケールアウト (Option) SQL Server Big Data Cluster Apache Hadoop YARN クエリエンジン SQL Engine Spark Engine 実行計画内部管理: Cost base Optimizer Spark ベース: Spark 実行プランクエリ実行 SQL Engine 内部処理 Executor による Scala - Java ベースの実行機械学習 SQL Server Machine Learning service – R / Python SparkML グラフクエリ SQL Graph GraphX 管理ツール SQL Server Management Studio | Azure Data Studio Spark UI 開発言語 ODBC/OLEDB/JDBC など各種ライブラリーから多くの開発言語に対応 Python / R / Spark SQL / Scala / Java / .NET

Spark Instance VM – 001 Node Agent Hive Metastore YARN RM - 01 Zookeeper - 01 Livy - 01 VM – 002 Node Agent YARN RM - 02 Zookeeper - 02 VM – 003 Node Agent YARN NM - 03 Zookeeper - 03 VM – 004 Node Agent YARN NM - 04 Subnet VM – 005 Node Agent Synapse Cluster Service (Control Plane) Heartbeat sequence Azure Resource Provider Create VMs with Specialized VHD Provision Resources Heartbeats Create Cluster YARN NM - 02 YARN NM - 01 Spark Executors Spark Executors Spark Executors Spark Executors

Synapse Service Job Service Frontend Spark API Controller … Gateway Resource Provider DB Synapse Studio AAD Auth Service Instance Creation Service DB Azure Job Service Backend Spark Plugin DB … Spark Instance VM VM VM VM VM VM

デフォルト設定必須項目

component versions, auto-pause PyPi (pip freeze) のファイルフォーマット。一般的には requirements.txt

In the Portal Specify the new requirements while creating Spark Pool in Additional Settings blade

PIP requirements file standard

Collect Data ファイルの読み取り SQL Databases へのクエリ Web Services 呼び出し Web Pages のスクレイプ Prepare Data データの探索データの妥当性検証データのクリーンアップデータの特徴理解 Train Model 学習用データセット作成実験環境学習の評価学習状況の可視化 Evaluate Model モデルのテストモデルの比較モデルの妥当性検証評価状況の可視化 Deploy Model モデルのエクスポートジョブの準備コンテナのデプロイ再パッケージング Notebook (Synapse Spark) 実行可能なもの: - Python code - R code - SQL queries - Shell commands

コードかテキストの追加コードもしくはテキストの削除順番の変更

Magic Command 説明 %%pyspark Python %%spark Scala %%sql Spark SQL %%csharp Spark.NET C# %%time セルの実行時間 %%capture stdout, stderror %%writefile セルの内容をファイルに出力 Cell の Magic Command セル単位で実行環境を設定する最初の行に記載する必要がありいずれも Spark 環境での動作

Toolbar からの実行選択しているセルから

Toolbar から出力結果の削除セルとして出力

保存 (publishing) 出力 (ダウンロード .ipynb ファイル )

Pipeline のステップの一つとしてpipeline activities

SQL Pool の table を DataFrame に直接読み込める

2 Analytics runtimes SQL Common data estate Shared meta data Unified experience Synapse Studio Store Azure Data Lake Storage Gen2 Power BI Azure Machine Learning クラウドデータ SaaS データオンプレミスデータデバイスデータ

…これらを試してみる • Spark で事前処理 • 複雑な CSV の読み込みに Polybase ではなく、COPY コマンドを試してみる

Analytics Runtimes SQL serverless SQL Runtime 共有クラスター自動スケール共有ストレージ SQL pool SQL Runtime プロビジョンクラスター現在、手動スケールのみ共有ストレージ専用ストレージクラスター × N 個を作成可能クラスター × N 個を作成可能 Spark pool Spark Runtime プロビジョンクラスター自動/手動スケール共有ストレージデータ統合機能パイプライン/データフロー Azure Data Lake Storage Gen2 ※現在、Spark pool から SQL Analytics にメタデータ共有可能なのは、Parquet ファイル形式のマネージドテーブル/外部テーブルで、SQL pool への共有は Synapse Studio 開発/監視/管理

• リソース状況認識型タスクスケジューラーによる高い同時実行性 • 交換理性:自己チューニング型のワークロードマネジメント：キャパシティプランニング, アクセス制御, リソースガバナンス Azure Storage Synapse SQL query service Power BI Azure Data Studio SSMS Read and write data files Sync table definitions Synapse Spark pool

Spark Table 作成 # ADLS からファイルを読み込む df = spark.read.load('abfss://wwi- 02@asadatalakedahatake.dfs.core.windows.net/sale- small/Year=2019/Quarter=Q1/Month=1/Day=20190101/sale-small-20190101- snappy.parquet', format='parquet') display(df.limit(10)) # Spark Table に保存 df.write.mode("overwrite").saveAsTable("default.salessmall") # クエリ実行 df = spark.sql("SELECT * FROM default.salessmall") df.show(10)

val jdbcUsername = "<SQL DB ADMIN USER>" val jdbcPwd = "<SQL DB ADMIN PWD>" val jdbcHostname = "servername.database.windows.net” val jdbcPort = 1433 val jdbcDatabase ="<AZURE SQL DB NAME>“ val jdbc_url = s"jdbc:sqlserver://${jdbcHostname}:${jdbcPort};database =${jdbcDatabase};encrypt=true;trustServerCertificate=fa lse;hostNameInCertificate=*.database.windows.net;loginT imeout=60;“ val connectionProperties = new Properties() connectionProperties.put("user", s"${jdbcUsername}") connectionProperties.put("password", s"${jdbcPwd}") val sqlTableDf = spark.read.jdbc(jdbc_url, “dbo.Tbl1", connectionProperties) // Construct a Spark DataFrame from SQL Pool table var df = spark.read.sqlanalytics("sql1.dbo.Tbl1") // Write the Spark DataFrame into SQL Pool table df.write.sqlanalytics(“sql1.dbo.Tbl2”) 従来型 Synapse での Scala %%spark val df = spark.read.sqlanalytics("sql1.dbo.Tbl1") df.createOrReplaceTempView("tbl1") %%pyspark sample = spark.sql("SELECT * FROM tbl1") sample.createOrReplaceTempView("tblnew") %%spark val df = spark.sql("SELECT * FROM tblnew") df.write.sqlanalytics(“sql1.dbo.tbl2", Constants.INTERNAL) Synapse での Python

取り込み、加工、可視化の原則は変わらない Store Transform QueryIngest Azure Data Lake Storage Gen2 SQL Pool SQL serverless クラウドデータ SaaS データオンプレミスデータデバイスデータ Power BI Azure Machine Learning

探検理解小さなサイズ信頼しない • 早めに失敗外に

だけ嫌い結果視野を広げて

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

Step-by-Step Achievements スムーズな学習環境  無料  日本語対応  ブラウザーのみでOK ハンズオン環境も含めて  ダウンロード可能なサンプルコード  Product/Service, 技術レベル, job role, などに応じたガイダンス  Video, チュートリアル, ハンズオン  スキルアップを促す  ユーザープロファイル毎にカスタマイズ www.microsoft.com/learn

www.microsoft.com/ja-jp/events

SQL Server 使いのための Azure Synapse Analytics - Spark 入門

More Related Content

What's hot

Similar to SQL Server 使いのための Azure Synapse Analytics - Spark 入門

More from Daiyu Hatakeyama

Recently uploaded

SQL Server 使いのための Azure Synapse Analytics - Spark 入門