チュートリアル:Azure HDInsight を使用してデータの抽出、変換、読み込みを行う

[アーティクル]
03/16/2023

このチュートリアルでは、ETL (データの抽出、変換、読み込み) 操作を実行します。生の CSV データファイルを取得して Azure HDInsight クラスターにインポートした後、Apache Hive を使用して変換し、Apache Sqoop を使用して Azure SQL Database に読み込みます。

このチュートリアルでは、以下の内容を学習します。

データを抽出し、HDInsight クラスターにアップロードする。
Apache Hive を使用してデータを変換する。
Sqoop を使用して Azure SQL Database にデータを読み込む。

Azure サブスクリプションをお持ちでない場合は、開始する前に無料アカウントを作成してください。

前提条件

HDInsight 用に構成された階層型名前空間 (Azure Data Lake Storage Gen2) を持つストレージアカウント

「Use Azure Data Lake Storage Gen2 with Azure HDInsight clusters」 (Azure HDInsight クラスターで Azure Data Lake Storage Gen2 を使用する) を参照してください。
HDInsight での Linux ベースの Hadoop クラスター

「クイックスタート:Azure portal を使用して Azure HDInsight で Apache Hadoop と Apache Hive を使用する」を参照してください。
Azure SQL Database

保存先データストアとして Azure SQL Database を使用します。 SQL Database のデータベースがない場合は、Azure portal での Azure SQL Database のデータベースの作成に関するページを参照してください。
Azure CLI

Azure CLI をインストールしていない場合は、「Azure CLI のインストール」を参照してください。
Secure Shell (SSH) クライアント

詳細については、SSH を使用した HDInsight (Hadoop) への接続に関するページを参照してください。

データをダウンロードおよび抽出してアップロードする

このセクションでは、サンプルフライトデータをダウンロードします。その後、HDInsight クラスターにデータをアップロードしてから、そのデータを自分の Data Lake Storage Gen2 アカウントにコピーします。

On_Time_Reporting_Carrier_On_Time_Performance_1987_present_2016_1.zip ファイルをダウンロードします。このファイルには、フライトデータが含まれています。
コマンドプロンプトを開き、次の Secure Copy (Scp) コマンドを使用して HDInsight クラスターのヘッドノードに .ZIP ファイルをアップロードします。
```
scp On_Time_Reporting_Carrier_On_Time_Performance_1987_present_2016_1.zip <ssh-user-name>@<cluster-name>-ssh.azurehdinsight.net:
```
- <ssh-user-name> プレースホルダーを HDInsight クラスターの SSH ユーザー名に置き換えます。
- <cluster-name> プレースホルダーを HDInsight クラスターの名前に置き換えます。
パスワードを使用して SSH ユーザー名を認証する場合は、パスワードを入力するよう求められます。

公開キーを使用している場合は、-i パラメーターを使用して、対応する秘密キーへのパスを指定することが必要な場合があります。たとえば、「 scp -i ~/.ssh/id_rsa <file_name>.zip <user-name>@<cluster-name>-ssh.azurehdinsight.net: 」のように入力します。
アップロードが完了したら、SSH を使用してクラスターに接続します。コマンドプロンプトで次のコマンドを入力します。
```
ssh <ssh-user-name>@<cluster-name>-ssh.azurehdinsight.net
```
次のコマンドを使用して .zip ファイルを解凍します。
```
unzip <file-name>.zip
```
このコマンドにより .csv ファイルが抽出されます。
次のコマンドを使用して、Data Lake Storage Gen2 コンテナーを作成します。
```
hadoop fs -D "fs.azure.createRemoteFileSystemDuringInitialization=true" -ls abfs://<container-name>@<storage-account-name>.dfs.core.windows.net/
```
<container-name> プレースホルダーを、ご自身のコンテナーに付ける名前に置き換えます。

<storage-account-name> プレースホルダーは、実際のストレージアカウントの名前に置き換えます。

次のコマンドを使用して、ディレクトリを作成します。

hdfs dfs -mkdir -p abfs://<container-name>@<storage-account-name>.dfs.core.windows.net/tutorials/flightdelays/data

次のコマンドを使用して .csv ファイルをディレクトリにコピーします。
```
hdfs dfs -put "On_Time_Reporting_Carrier_On_Time_Performance_(1987_present)_2016_1.csv" abfs://<container-name>@<storage-account-name>.dfs.core.windows.net/tutorials/flightdelays/data/
```
ファイル名にスペースや特殊文字が含まれる場合は、ファイル名を引用符で囲んでください。

データの変換

このセクションでは、Beeline を使用して Apache Hive ジョブを実行します。

Apache Hive ジョブの一環として、.csv ファイルから delays という名前の Apache Hive テーブルにデータをインポートします。

HDInsight クラスター用に既に開いている SSH プロンプトから、次のコマンドを使用して flightdelays.hql という名前の新しいファイルを作成して編集します。
```
nano flightdelays.hql
```

次のテキストに変更を加えます。<container-name> と <storage-account-name> のプレースホルダーを実際のコンテナーとストレージアカウントの名前に置き換えてください。マウスの右ボタンを選択したまま Shift キーを押して、そのテキストを nano コンソールにコピーして貼り付けます。

  DROP TABLE delays_raw;
  -- Creates an external table over the csv file
  CREATE EXTERNAL TABLE delays_raw (
     YEAR string,
     FL_DATE string,
     UNIQUE_CARRIER string,
     CARRIER string,
     FL_NUM string,
     ORIGIN_AIRPORT_ID string,
     ORIGIN string,
     ORIGIN_CITY_NAME string,
     ORIGIN_CITY_NAME_TEMP string,
     ORIGIN_STATE_ABR string,
     DEST_AIRPORT_ID string,
     DEST string,
     DEST_CITY_NAME string,
     DEST_CITY_NAME_TEMP string,
     DEST_STATE_ABR string,
     DEP_DELAY_NEW float,
     ARR_DELAY_NEW float,
     CARRIER_DELAY float,
     WEATHER_DELAY float,
     NAS_DELAY float,
     SECURITY_DELAY float,
     LATE_AIRCRAFT_DELAY float)
  -- The following lines describe the format and location of the file
  ROW FORMAT DELIMITED FIELDS TERMINATED BY ','
  LINES TERMINATED BY '\n'
  STORED AS TEXTFILE
  LOCATION 'abfs://<container-name>@<storage-account-name>.dfs.core.windows.net/tutorials/flightdelays/data';

  -- Drop the delays table if it exists
  DROP TABLE delays;
  -- Create the delays table and populate it with data
  -- pulled in from the CSV file (via the external table defined previously)
  CREATE TABLE delays
  LOCATION 'abfs://<container-name>@<storage-account-name>.dfs.core.windows.net/tutorials/flightdelays/processed'
  AS
  SELECT YEAR AS year,
     FL_DATE AS FlightDate, 
     substring(UNIQUE_CARRIER, 2, length(UNIQUE_CARRIER) -1) AS IATA_CODE_Reporting_Airline,
     substring(CARRIER, 2, length(CARRIER) -1) AS Reporting_Airline, 
     substring(FL_NUM, 2, length(FL_NUM) -1) AS Flight_Number_Reporting_Airline,
     ORIGIN_AIRPORT_ID AS OriginAirportID,
     substring(ORIGIN, 2, length(ORIGIN) -1) AS OriginAirportSeqID,
     substring(ORIGIN_CITY_NAME, 2) AS OriginCityName,
     substring(ORIGIN_STATE_ABR, 2, length(ORIGIN_STATE_ABR) -1)  AS OriginState,
     DEST_AIRPORT_ID AS DestAirportID,
     substring(DEST, 2, length(DEST) -1) AS DestAirportSeqID,
     substring(DEST_CITY_NAME,2) AS DestCityName,
     substring(DEST_STATE_ABR, 2, length(DEST_STATE_ABR) -1) AS DestState,
     DEP_DELAY_NEW AS DepDelay,
     ARR_DELAY_NEW AS ArrDelay,
     CARRIER_DELAY AS CarrierDelay,
     WEATHER_DELAY AS WeatherDelay,
     NAS_DELAY AS NASDelay,
     SECURITY_DELAY AS SecurityDelay,
     LATE_AIRCRAFT_DELAY AS LateAircraftDelay
  FROM delays_raw;

Ctrl キーを押しながら X キーを押し、確認を求められたら「Y」と入力してファイルを保存します。
Hive を起動して flightdelays.hql ファイルを実行するには、次のコマンドを使います。
```
beeline -u 'jdbc:hive2://localhost:10001/;transportMode=http' -f flightdelays.hql
```
flightdelays.hql スクリプトの実行が終わったら、次のコマンドで対話型 Beeline セッションを開始します。
```
beeline -u 'jdbc:hive2://localhost:10001/;transportMode=http'
```
jdbc:hive2://localhost:10001/> プロンプトが表示されたら、次のクエリを使用してインポートされたフライト遅延データからデータを取得します。
```
INSERT OVERWRITE DIRECTORY '/tutorials/flightdelays/output'
ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t'
SELECT regexp_replace(OriginCityName, '''', ''),
  avg(WeatherDelay)
FROM delays
WHERE WeatherDelay IS NOT NULL
GROUP BY OriginCityName;
```
このクエリにより、悪天候による遅延が発生した都市の一覧と平均遅延時間が取得され、abfs://<container-name>@<storage-account-name>.dfs.core.windows.net/tutorials/flightdelays/output に保存されます。その後、Sqoop がこの場所からデータを読み取り、Azure SQL Database にエクスポートします。
Beeline を終了するには、プロンプトで「 !quit 」と入力します。

SQL データベーステーブルの作成

この操作には、SQL Database のサーバー名が必要になります。サーバー名を確認するには、次の手順に従います。

Azure ポータルにアクセスします。
[SQL データベース] を選択します。
使用するデータベースの名前でフィルター処理します。サーバー名は [サーバー名] 列に表示されます。
使用するデータベースの名前でフィルター処理します。サーバー名は [サーバー名] 列に表示されます。

SQL Database に接続してテーブルを作成するには、多くの方法があります。次の手順では、HDInsight クラスターから FreeTDS を使用します。
FreeTDS をインストールするには、クラスターへの SSH 接続から次のコマンドを使用します。
```
sudo apt-get --assume-yes install freetds-dev freetds-bin
```
インストールが完了したら、次のコマンドを使用して SQL Database に接続します。
```
TDSVER=8.0 tsql -H '<server-name>.database.windows.net' -U '<admin-login>' -p 1433 -D '<database-name>'
```
- <server-name> プレースホルダーを論理 SQL サーバーの名前に置き換えます。
- <admin-login> プレースホルダーを SQL Database の管理者ユーザー名に置き換えます。
- <database-name> プレースホルダーをデータベース名に置き換えます。
メッセージが表示されたら、SQL Database 管理者ユーザー名のパスワードを入力します。

次のテキストのような出力が返されます。
```
locale is "en_US.UTF-8"
locale charset is "UTF-8"
using default charset "UTF-8"
Default database being set to sqooptest
1>
```

1> プロンプトで、次のステートメントを入力します。

CREATE TABLE [dbo].[delays](
[OriginCityName] [nvarchar](50) NOT NULL,
[WeatherDelay] float,
CONSTRAINT [PK_delays] PRIMARY KEY CLUSTERED
([OriginCityName] ASC))
GO

GO ステートメントを入力すると、前のステートメントが評価されます。

このクエリにより、クラスター化インデックス付きの、delays という名前のテーブルが作成されます。

次のクエリを使用して、テーブルが作成されたことを確認します。

SELECT * FROM information_schema.tables
GO

出力は次のテキストのようになります。

TABLE_CATALOG   TABLE_SCHEMA    TABLE_NAME      TABLE_TYPE
databaseName       dbo             delays        BASE TABLE

Enter exit at the 1>」と入力して、tsql ユーティリティを終了します。

データのエクスポートと読み込み

上のセクションで、変換したデータを abfs://<container-name>@<storage-account-name>.dfs.core.windows.net/tutorials/flightdelays/output にコピーしました。このセクションでは、Sqoop を使用して、abfs://<container-name>@<storage-account-name>.dfs.core.windows.net/tutorials/flightdelays/output のデータを、Azure SQL Database に作成したテーブルにエクスポートします。

次のコマンドを使用して、Sqoop が SQL データベースを認識できることを確認します。
```
sqoop list-databases --connect jdbc:sqlserver://<SERVER_NAME>.database.windows.net:1433 --username <ADMIN_LOGIN> --password <ADMIN_PASSWORD>
```
このコマンドにより、delays テーブルを作成したデータベースを含む、データベースの一覧が返されます。
次のコマンドを使って、hivesampletable テーブルから delays テーブルにデータをエクスポートします。
```
sqoop export --connect 'jdbc:sqlserver://<SERVER_NAME>.database.windows.net:1433;database=<DATABASE_NAME>' --username <ADMIN_LOGIN> --password <ADMIN_PASSWORD> --table 'delays' --export-dir 'abfs://<container-name>@.dfs.core.windows.net/tutorials/flightdelays/output' --fields-terminated-by '\t' -m 1
```
Sqoop は delays テーブルを含むデータベースに接続して、/tutorials/flightdelays/output ディレクトリから delays テーブルにデータをエクスポートします。

sqoop コマンドが完了したら、tsql ユーティリティを使用してデータベースに接続します。

TDSVER=8.0 tsql -H <SERVER_NAME>.database.windows.net -U <ADMIN_LOGIN> -P <ADMIN_PASSWORD> -p 1433 -D <DATABASE_NAME>

次のステートメントを使用して、データが delays テーブルにエクスポートされたことを確認します。
```
SELECT * FROM delays
GO
```
テーブル内のデータの一覧が表示されます。テーブルには、都市の名前と、その都市のフライトの平均遅延時間が含まれます。
「exit」と入力して、tsql ユーティリティを終了します。

リソースをクリーンアップする

このチュートリアルで使用されるすべてのリソースは、既存のものです。クリーンアップは必要ありません。

次のステップ

HDInsight でのデータ操作の詳細については、次の記事を参照してください。

Azure HDInsight クラスターで Azure Data Lake Storage Gen2 を使用する

次の方法で共有

チュートリアル:Azure HDInsight を使用してデータの抽出、変換、読み込みを行う

前提条件

データをダウンロードおよび抽出してアップロードする

データの変換

SQL データベーステーブルの作成

データのエクスポートと読み込み

リソースをクリーンアップする

次のステップ

フィードバック

その他のリソース

次の方法で共有

チュートリアル:Azure HDInsight を使用してデータの抽出、変換、読み込みを行う

前提条件

データをダウンロードおよび抽出してアップロードする

データの変換

SQL データベース テーブルの作成

データのエクスポートと読み込み

リソースをクリーンアップする

次のステップ

フィードバック

その他のリソース

SQL データベーステーブルの作成