やましーデータ活用クラウドエンジニアになるSQLも叩けない、クラウドも触ったことがない三十代元インフラエンジニアの私。2020年1月16日こんにちは。Learnの続きです。全3回のうちの3回目記事になります。FactoryのパイプラインからDatabricksを呼び出し、Databricksの中でデータを整形してデータを統合します。そのあとDatabricksから統合したデータをグラフ化します。ネタバレですがこんなグラフを出力します。Databricksで犯罪データをグラフ化した3回の記事を通じて、データをソースからデータ基盤に持ってきて、分散処理基盤(Databricks)によりデータを整形し、可視化することができるようになります。今回の記事を読んでクラウドを使いこなせるようになって憧れのつよつよエンジニアになってください。初回、2回目の記事の中で構築した環境を引き続き使うのでまだ読んでないよって方はリンクを貼っておきますのでどうぞ。目次を見るだけでも何をしているかが分かると思います。Factoryを連携させたいそれでは始めていきます。DatabricksにアクセスしてNotebookの処理を実行させるためにはDatabricksへのアクセストークンを取得する必要があります。Databriksのワークスペースへ移動します。移動したらNotebookを開いていきます。1人のアイコンを選択します。1「UserSettings」を選択します。Token」を選択します。access」わかりやすいコメントを付けます。(days):「90」アクセストークンの有効期間を指定します。3「Generate」を選択します。1アクセストークンの文字列をコピーして控えておきます。ここでコピーをしてください。後から参照することはできません。2「Done」を選択します。Factory管理画面から「作成と監視」を選択します。1鉛筆のアイコンを選択します。1Pipelineから前回作成した「LabPipeline」を選択します。2Activeitiesの中からDatabricksを開き「Notebook」をドラッグ&ドロップします。dataから出ている緑のタブを伸ばしてNoteBookに接続します。1Name:「LabNotebook」としました。serviceにDatabricksのクラスタを追加していきます。Databricks」を選択します。2「+New」を選択し新規追加します。service名を設定します。runtime:「AutoResolveintegrationRuntime」ランタイムの種類はAzureマネージドなものを選択しました。使用するDatabricksのワークスペースをサブスクリプションから選択します。workspace:「ymstech-adb01」前回作成したワークスペースを選択します。作成済みのクラスターを選択します。token:先ほど生成してコピーしておいたアクセストークンの文字列を設定します。clusters:「ymstech-cluster01」を設定します。connection」を選択し接続を確認します。successful」となることを確認します。11「Create」を選択します。パイプラインで実行するNoteBookを選択するパイプラインで実行するNotebookを選択していきます。このNotebookはストレージへコピーした犯罪データの各データでバラバラな列名を統一しデータを統合する処理が記載されています。1「Settings」を選択します。Path:「/Users/[ユーザー名]/03-Data-Ingestion-Via-ADF/includes/Databricks-Data-Transformations」Browseを選択して選ぶこともできます。今回はすでにインポート済のNotebookを使います。3「+New」を選択します。今回のNotebookは引数を期待するので入力します。4初回で作成したストレージアカウントの情報を設定します。パラメータを取るNotebookの書き方1Notebookはwidgetsメソッドでパラメータを取ることができるようになります。指定した変換Notebookを見てみます。uppercol()は、指定された列名に基づいて、DataFrameから列を返します。contains(mySubstr)は、サブストリングmySubstrを含むストリングを示します。lit()を使用して、リテラル値から列を作成します。データが派生する都市の名前を含む各DataFrameに新しい列を作成します。lower()は、テキストを小文字に変換します。month()は、reportDateタイムスタンプデータ型から月を抽出します。unix_timestamp()を使用して、ダラスの日付フィールドをタイムスタンプ形式に変換します。したがって、month()関数を使用して月を抽出できます。homicidesNewYorkDf.union(homicidesBostonDf).union(homicidesChicagoDf).union(homicidesDallasDf).union(homicidesLosAngelesDf).union(homicidesNewOrleansDf).union(homicidesPhillyDf)homicidesDf.write.mode("overwrite").saveAsTable("homicides_2016")ちなみにテーブルセーブすると以下の場所から参照することができます。1「Data」を選択します。2Databasesから「Default」を選択します。テーブルを選択します。4概要が表示されます。最後にNotebookの戻り値を返す処理を記述しています。["homicides_2016"]}))今回DataFactoryのパイプラインから使用する変換処理のNotebookは以上です。all」を実行し、確定させます。completed」となることを確認します。これでパイプラインが確定し使える状態になりました。Databricksを使ってデータを統合します。パイプラインを実行するパイプラインを実行していきます。trigger」を選択します。now」を選択します。今回は手動で実行します。このほかにも時刻などのスケジュールによる起動、BLOBを監視してファイル作成/削除による起動、タンブリングウィンドウによる謎の起動方法などが選択できます。1今回のパイプラインにはパラメータはありませんので何も設定せず「Finish」を選択します。これでパイプラインが実行されました。パイプラインの実行結果を確認するパイプラインの実行結果を確認してみます。1「モニターのアイコン」を選択します。runs」を選択し、パイプラインの実行結果を見てみます。32つのアクティビティが「Succeeded」となっていることが分かります。次により詳細なNotebookの実行結果を見ていきたいと思います。DatabricksのNotebookの実行結果を確認する1DatabricksNotebookアクティビティのアウトプットのアイコンを選択します。2runPageUrlを選択します。実行結果のページに遷移します。各コマンドの結果を確認することができます。今回テーブルを保存した際にアテンションが表示されていました。Lake」と呼ばれるフォーマットがリリースされ、こちらにすることでパフォーマンスを向上できることを促しているようです。Lakeフォーマットはテーブルを差分管理することによりロールバック等タイムトラベルが行えます。ZORDERによる最適化を行うことによりパフォーマンス向上が見込めます。統合したテーブルをグラフ化するいよいよ大詰めですね。統合したテーブルをグラフ化していきたいと思います。