Kurz: Extrakce, transformace a načítání dat pomocí Azure HDInsightu

V tomto kurzu provedete operaci ETL: extrakci, transformaci a načtení dat. Vezmete nezpracovaný datový soubor CSV, naimportujete ho do clusteru Azure HDInsight, transformujete ho pomocí Apache Hivu a načtete ho do Azure SQL Database pomocí Apache Sqoopu.

V tomto kurzu se naučíte:

• Extrahujte a nahrajte data do clusteru HDInsight.
• Transformujte data pomocí Apache Hivu.
• Načtěte data do služby Azure SQL Database pomocí Sqoopu.

Pokud ještě nemáte předplatné Azure, vytvořte si napřed bezplatný účet.

Požadavky

• Účet úložiště, který má hierarchický obor názvů (Azure Data Lake Storage), který je nakonfigurovaný pro HDInsight

  Viz Použití služby Azure Data Lake Storage s clustery Azure HDInsight.

• Cluster Hadoop založený na Linuxu ve službě HDInsight

  Viz rychlý start: Začínáme s Apache Hadoopem a Apache Hivem ve službě Azure HDInsight pomocí webu Azure Portal.

• Azure SQL Database

  Azure SQL Database použijete jako cílové úložiště dat. Pokud databázi nemáte ve službě SQL Database, přečtěte si téma Vytvoření databáze ve službě Azure SQL Database na webu Azure Portal.

• Azure CLI

  Pokud jste ještě nenainstalovali Azure CLI, přečtěte si téma Instalace Azure CLI.

• Klient SSH (Secure Shell)

  Další informace najdete v tématu Připojení ke službě HDInsight (Hadoop) pomocí SSH.

Stažení, extrakce a nahrání dat

V této části si stáhnete ukázková testovací data. Potom tato data nahrajete do clusteru HDInsight a pak tato data zkopírujete do svého účtu Data Lake Storage.

1. Stáhněte soubor On_Time_Reporting_Carrier_On_Time_Performance_1987_present_2016_1.zip. Tento soubor obsahuje letová data.

2. Otevřete příkazový řádek a pomocí následujícího příkazu SCP (Secure Copy) nahrajte soubor .zip do hlavního uzlu clusteru HDInsight:

   scp On_Time_Reporting_Carrier_On_Time_Performance_1987_present_2016_1.zip <ssh-user-name>@<cluster-name>-ssh.azurehdinsight.net:
   

   • <ssh-user-name> Zástupný symbol nahraďte uživatelským jménem SSH pro cluster HDInsight.
   • <cluster-name> Zástupný symbol nahraďte názvem clusteru HDInsight.

   Pokud k ověření uživatelského jména SSH použijete heslo, zobrazí se výzva k zadání hesla.

   Pokud používáte veřejný klíč, budete pravděpodobně muset použít parametr -i k zadání cesty k odpovídajícímu privátnímu klíči. Například scp -i ~/.ssh/id_rsa <file_name>.zip <user-name>@<cluster-name>-ssh.azurehdinsight.net:.

3. Po dokončení nahrávání se ke clusteru připojte pomocí SSH. Na příkazovém řádku zadejte následující příkaz:

   ssh <ssh-user-name>@<cluster-name>-ssh.azurehdinsight.net
   

4. Pomocí následujícího příkazu rozbalte soubor .zip:

   unzip <file-name>.zip
   

   Příkaz extrahuje soubor .csv .

5. Pomocí následujícího příkazu vytvořte kontejner Data Lake Storage.

   hadoop fs -D "fs.azure.createRemoteFileSystemDuringInitialization=true" -ls abfs://<container-name>@<storage-account-name>.dfs.core.windows.net/
   

   <container-name> Zástupný symbol nahraďte názvem, který chcete kontejneru udělit.

   <storage-account-name> Zástupný symbol nahraďte názvem vašeho účtu úložiště.

6. K vytvoření adresáře použijte následující příkaz.

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

7. Pomocí následujícího příkazu zkopírujte soubor .csv do adresáře:

   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/
   

   Pokud název souboru obsahuje mezery nebo speciální znaky, použijte uvozovky kolem názvu souboru.

Transformace dat

V této části použijete Beeline ke spuštění úlohy Apache Hive.

V rámci úlohy Apache Hive importujete data ze souboru .csv do tabulky Apache Hive s názvem zpoždění.

1. Na příkazovém řádku SSH, který už máte pro cluster HDInsight spuštěný, pomocí následujícího příkazu vytvořte a upravte nový soubor flightdelays.hql:

   nano flightdelays.hql
   

2. Upravte následující text nahrazením <container-name> a <storage-account-name> zástupnými symboly názvem kontejneru a účtu úložiště. Potom text zkopírujte a vložte do konzoly Nano stisknutím klávesy SHIFT spolu s tlačítkem pro výběr pravého tlačítka myši.

     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;
   

3. Uložte soubor zadáním ctrl+X a zadáním Y po zobrazení výzvy.

4. Pokud chcete spustit Hive a spustit flightdelays.hql soubor, použijte následující příkaz:

   beeline -u 'jdbc:hive2://localhost:10001/;transportMode=http' -f flightdelays.hql
   

5. flightdelays.hql Po dokončení spuštění skriptu pomocí následujícího příkazu otevřete interaktivní relaci Beeline:

   beeline -u 'jdbc:hive2://localhost:10001/;transportMode=http'
   

6. Po zobrazení příkazového řádku jdbc:hive2://localhost:10001/> pomocí následujícího dotazu načtěte data z importovaných dat o zpožděných letech:

   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;
   

   Tento dotaz načte seznam měst, ve kterých došlo ke zpožděním kvůli nepřízni počasí, společně s průměrnou délkou zpoždění a uloží ho do umístění abfs://<container-name>@<storage-account-name>.dfs.core.windows.net/tutorials/flightdelays/output. Později z tohoto umístění data načte Sqoop a exportuje je do služby Azure SQL Database.

7. Beeline ukončíte zadáním !quit na příkazovém řádku.

Vytvoření tabulky databáze SQL

Pro tuto operaci potřebujete název serveru ze služby SQL Database. Pokud chcete najít název serveru, proveďte tyto kroky.

1. Přejděte na Azure Portal.

2. Vyberte databáze SQL.

3. Vyfiltrujte název databáze, kterou chcete použít. Název serveru se zobrazí ve sloupci Název serveru.

4. Vyfiltrujte název databáze, kterou chcete použít. Název serveru se zobrazí ve sloupci Název serveru.

   

   Existuje mnoho způsobů, jak se připojit ke službě SQL Database a vytvořit tabulku. V následujících krocích se používá FreeTDS z clusteru HDInsight.

5. Pokud chcete nainstalovat FreeTDS, použijte následující příkaz z připojení SSH ke clusteru:

   sudo apt-get --assume-yes install freetds-dev freetds-bin
   

6. Po dokončení instalace se pomocí následujícího příkazu připojte ke službě SQL Database.

   TDSVER=8.0 tsql -H '<server-name>.database.windows.net' -U '<admin-login>' -p 1433 -D '<database-name>'
   

   • <server-name> Zástupný text nahraďte názvem logického serveru SQL.

   • <admin-login> Zástupný symbol nahraďte uživatelským jménem správce pro SLUŽBU SQL Database.

   • Nahrazení zástupného <database-name> symbolu názvem databáze

   Po zobrazení výzvy zadejte heslo pro uživatelské jméno správce služby SQL Database.

   Zobrazí se výstup podobný následujícímu textu:

   locale is "en_US.UTF-8"
   locale charset is "UTF-8"
   using default charset "UTF-8"
   Default database being set to sqooptest
   1>
   

7. Na příkazovém 1> řádku zadejte následující příkazy:

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

8. Po zadání příkazu GO se vyhodnotí předchozí příkazy.

   Dotaz vytvoří tabulku s názvem zpoždění, která má clusterovaný index.

9. Pomocí následujícího dotazu ověřte, že je tabulka vytvořená:

   SELECT * FROM information_schema.tables
   GO
   

   Výstup se bude podobat následujícímu:

   TABLE_CATALOG   TABLE_SCHEMA    TABLE_NAME      TABLE_TYPE
   databaseName       dbo             delays        BASE TABLE
   

10. Zadáním exit na příkazovém řádku 1> ukončete nástroj tsql.

Export a načtení dat

V předchozích částech jste zkopírovali transformovaná data v umístění abfs://<container-name>@<storage-account-name>.dfs.core.windows.net/tutorials/flightdelays/output. V této části použijete Sqoop k exportu dat z abfs://<container-name>@<storage-account-name>.dfs.core.windows.net/tutorials/flightdelays/output tabulky, kterou jste vytvořili ve službě Azure SQL Database.

1. Pomocí následujícího příkazu ověřte, že má Sqoop vhled do vaší databáze SQL:

   sqoop list-databases --connect jdbc:sqlserver://<SERVER_NAME>.database.windows.net:1433 --username <ADMIN_LOGIN> --password <ADMIN_PASSWORD>
   

   Příkaz vrátí seznam databází, včetně databáze, ve které jste vytvořili tabulku zpoždění .

2. K exportu dat z tabulky hivesampletable do tabulky zpoždění použijte následující příkaz:

   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 se připojí k databázi, která obsahuje tabulku zpoždění , a exportuje data z /tutorials/flightdelays/output adresáře do tabulky zpoždění .

3. sqoop Po dokončení příkazu se pomocí nástroje tsql připojte k databázi:

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

4. Pomocí následujících příkazů ověřte, že se data exportovala do tabulky zpoždění :

   SELECT * FROM delays
   GO
   

   Měl by se zobrazit výpis dat v tabulce. Tabulka obsahuje název města a průměrnou délku zpoždění letu pro příslušné město.

5. Zadáním ukončete exit nástroj tsql.

Vyčištění prostředků

Všechny prostředky použité v tomto kurzu už existují. Není nutné nic vyčistit.

Další kroky

Další způsoby práce s daty ve službě HDInsight najdete v následujícím článku:

Použití služby Azure Data Lake Storage s clustery Azure HDInsight