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

V tomto kurzu provedete operaci ETL: extrakce, transformace 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 Azure SQL Database pomocí Sqoopu.

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

Požadavky

• Účet úložiště s hierarchickým oborem názvů (Azure Data Lake Storage Gen2) nakonfigurovaným pro HDInsight

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

• Cluster Hadoop se systémem Linux ve službě HDInsight

  Viz Rychlý start: Začínáme s Apache Hadoopem a Apache Hivem ve službě Azure HDInsight s využitím Azure Portal.

• Azure SQL Database

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

• Azure CLI

  Pokud jste ještě azure CLI nenainstalovali, 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í, extrahování a nahrání dat

V této části si stáhnete ukázková letová data. Pak tato data nahrajete do clusteru HDInsight a zkopírujete je do účtu Data Lake Storage Gen2.

1. Stáhněte si souborOn_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 Secure Copy (Scp) 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ý text 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 Gen2.

   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 dát.

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

6. Pomocí následujícího příkazu vytvořte adresář.

   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.

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 delays.

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> zástupných symbolů a <storage-account-name> názvem vašeho kontejneru a účtu úložiště. Potom text zkopírujte a vložte do konzoly nano pomocí klávesy SHIFT a tlačítka pro výběr pravým tlačítkem 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 tak, že zadáte CTRL+X a po zobrazení výzvy zadáte Y text.

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 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 z SQL Database. Pomocí těchto kroků vyhledejte název vašeho serveru.

1. Přejděte na Azure Portal.

2. Vyberte Databáze SQL.

3. Vyfiltrujte název databáze, kterou se rozhodnete 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 k 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 logickým názvem serveru SQL.

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

   • <database-name> Nahraďte zástupný symbol názvem databáze.

   Po zobrazení výzvy zadejte heslo pro uživatelské jméno správce 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 delays, 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 do tabulky, kterou jste vytvořili v databázi Azure SQL.

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 delays 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 delays , a exportuje data z /tutorials/flightdelays/output adresáře do tabulky delays .

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 vyexportovala do tabulky delays :

   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 exit příkazu ukončete nástroj tsql.

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

Všechny prostředky použité v tomto kurzu už existují. Není nutné žádné čištění.

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 Gen2 s clustery Azure HDInsight