Öğretici: Azure HDInsight kullanarak verileri ayıklama, dönüştürme ve yükleme

Bu öğreticide bir ETL işlemi gerçekleştirirsiniz: verileri ayıklama, dönüştürme ve yükleme. Ham csv veri dosyasını alır, azure HDInsight kümesine aktarır, Apache Hive ile dönüştürür ve Apache Sqoop ile Azure SQL Veritabanına yüklersiniz.

Bu öğreticide şunların nasıl yapıldığını öğreneceksiniz:

• Verileri ayıklayın ve bir HDInsight kümesine yükleyin.
• Apache Hive kullanarak verileri dönüştürün.
• Sqoop kullanarak verileri Azure SQL Veritabanına yükleyin.

Azure aboneliğiniz yoksa başlamadan önce ücretsiz bir hesap oluşturun.

Önkoşullar

• HDInsight için yapılandırılmış hiyerarşik ad alanına (Azure Data Lake Storage 2. Nesil) sahip depolama hesabı

  Bkz. Azure HDInsight kümeleriyle Azure Data Lake Storage 2. Nesil kullanma.

• HDInsight üzerinde Linux tabanlı Hadoop kümesi

  Bkz. Hızlı Başlangıç: Azure portal kullanarak Azure HDInsight'ta Apache Hadoop ve Apache Hive kullanmaya başlama.

• Azure SQL Veritabanı

  Azure SQL Veritabanını hedef veri deposu olarak kullanırsınız. SQL Veritabanı'da veritabanınız yoksa bkz. Azure portal Azure SQL Veritabanında veritabanı oluşturma.

• Azure CLI

  Azure CLI'yi yüklemediyseniz bkz. Azure CLI'yi yükleme.

• Secure Shell (SSH) istemcisi

  Daha fazla bilgi için bkz. SSH kullanarak HDInsight'a (Hadoop) bağlanma.

Verileri indirme, ayıklama ve karşıya yükleme

Bu bölümde örnek uçuş verilerini indirirsiniz. Ardından bu verileri HDInsight kümenize yükler ve ardından bu verileri Data Lake Storage 2. Nesil hesabınıza kopyalarsınız.

1. On_Time_Reporting_Carrier_On_Time_Performance_1987_present_2016_1.zip dosyasını indirin. Bu dosya, uçuş verilerini içerir.

2. Bir komut istemi açın ve .zip dosyasını HDInsight kümesi baş düğümüne yüklemek için aşağıdaki Güvenli Kopya (Scp) komutunu kullanın:

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

   • Yer tutucuyu <ssh-user-name> HDInsight kümesinin SSH kullanıcı adıyla değiştirin.
   • Yer tutucuyu <cluster-name> HDInsight kümesinin adıyla değiştirin.

   SSH kullanıcı adınızın kimliğini doğrulamak için parola kullanırsanız parola istenir.

   Ortak anahtar kullanıyorsanız, eşleşen özel anahtarın yolunu belirtmek için -i parametresini kullanmanız gerekebilir. Örneğin, scp -i ~/.ssh/id_rsa <file_name>.zip <user-name>@<cluster-name>-ssh.azurehdinsight.net:.

3. Karşıya yükleme tamamlandıktan sonra SSH kullanarak kümeye bağlanın. Komut istemine aşağıdaki komutu girin:

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

4. .zip dosyasını açmak için aşağıdaki komutu kullanın:

   unzip <file-name>.zip
   

   komutu bir .csv dosyası ayıklar.

5. Data Lake Storage 2. Nesil kapsayıcısını oluşturmak için aşağıdaki komutu kullanın.

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

   Yer tutucuyu <container-name> kapsayıcınıza vermek istediğiniz adla değiştirin.

   Yer tutucuyu <storage-account-name> depolama hesabınızın adıyla değiştirin.

6. Dizin oluşturmak için aşağıdaki komutu kullanın.

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

7. .csv dosyasını dizine kopyalamak için aşağıdaki komutu kullanın:

   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/
   

   Dosya adında boşluklar veya özel karakterler varsa, dosya adının çevresinde tırnak işaretleri kullanın.

Verileri dönüştürme

Bu bölümde, Bir Apache Hive işi çalıştırmak için Beeline kullanacaksınız.

Apache Hive işinin bir parçası olarak, .csv dosyasındaki verileri delays adlı bir Apache Hive tablosuna aktarırsınız.

1. HDInsight kümesi için aldığınız SSH isteminden aşağıdaki komutu kullanarak flightdelays.hql adlı yeni bir dosya oluşturup düzenleyin:

   nano flightdelays.hql
   

2. ve <storage-account-name> yer tutucularını kapsayıcınız ve depolama hesabı adınızla değiştirerek <container-name> aşağıdaki metni değiştirin. Ardından, sağ fare seçme düğmesiyle birlikte SHIFT tuşuna basarak metni kopyalayıp nano konsola yapıştırın.

     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. CTRL+X tuşlarına basıp istendiğinde yazarak Y dosyayı kaydedin.

4. Hive'ı başlatmak ve dosyayı çalıştırmak flightdelays.hql için aşağıdaki komutu kullanın:

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

5. Betiğin flightdelays.hql çalışması tamamlandıktan sonra etkileşimli bir Beeline oturumu açmak için aşağıdaki komutu kullanın:

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

6. jdbc:hive2://localhost:10001/> istemini aldığınızda, içeri aktarılan uçuş gecikme verilerini almak için aşağıdaki sorguyu kullanın:

   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;
   

   Bu sorgu, hava durumundan kaynaklanan gecikmeler yaşayan şehirlerin bir listesini, ortalama gecikme süresi ile birlikte alır ve abfs://<container-name>@<storage-account-name>.dfs.core.windows.net/tutorials/flightdelays/output konumuna kaydeder. Daha sonra, Sqoop bu konumdaki verileri okur ve Azure SQL Veritabanına aktarır.

7. Beeline’dan çıkmak için isteme !quit girin.

SQL veritabanı tablosu oluşturma

Bu işlem için SQL Veritabanı sunucu adına ihtiyacınız var. Sunucu adınızı bulmak için bu adımları tamamlayın.

1. Azure Portal gidin.

2. SQL Veritabanları'ı seçin.

3. Kullanmayı seçtiğiniz veritabanının adına göre filtreleyin. Sunucu adı, Sunucu adı sütununda listelenir.

4. Kullanmak istediğiniz veritabanının adına göre filtreleyin. Sunucu adı, Sunucu adı sütununda listelenir.

   

   SQL Veritabanına bağlanıp tablo oluşturmanın çok sayıda yolu vardır. Aşağıdaki adımlarda HDInsight kümesinden FreeTDS kullanılır.

5. FreeTDS yüklemek için küme ile kurulan bir SSH bağlantısından aşağıdaki komutu kullanın:

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

6. Yükleme tamamlandıktan sonra, SQL Veritabanı bağlanmak için aşağıdaki komutu kullanın.

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

   • Yer tutucuyu <server-name> mantıksal SQL sunucusu adıyla değiştirin.

   • yer tutucuyu <admin-login> SQL Veritabanı için yönetici kullanıcı adıyla değiştirin.

   • Yer tutucuyu <database-name> veritabanı adıyla değiştirin

   İstendiğinde, SQL Veritabanı yönetici kullanıcı adının parolasını girin.

   Aşağıdakine benzer bir çıktı alırsınız:

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

7. İstemde 1> aşağıdaki deyimleri girin:

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

8. GO deyimi girildiğinde önceki deyimler değerlendirilir.

   Sorgu, kümelenmiş dizine sahip delays adlı bir tablo oluşturur.

9. Tablonun oluşturulduğunu doğrulamak için aşağıdaki sorguyu kullanın:

   SELECT * FROM information_schema.tables
   GO
   

   Çıktı aşağıdaki metne benzer:

   TABLE_CATALOG   TABLE_SCHEMA    TABLE_NAME      TABLE_TYPE
   databaseName       dbo             delays        BASE TABLE
   

10. Tsql yardımcı programından çıkış yapmak için 1> istemine exit girin.

Verileri dışarı aktarma ve yükleme

Önceki bölümlerde, dönüştürülen verileri konumunda abfs://<container-name>@<storage-account-name>.dfs.core.windows.net/tutorials/flightdelays/outputkopyalamıştınız. Bu bölümde Sqoop kullanarak verileri abfs://<container-name>@<storage-account-name>.dfs.core.windows.net/tutorials/flightdelays/output Azure SQL Veritabanında oluşturduğunuz tabloya aktaracaksınız.

1. Sqoop’un SQL veritabanınızı görebildiğini doğrulamak için aşağıdaki komutu kullanın:

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

   komut, delays tablosunu oluşturduğunuz veritabanı da dahil olmak üzere veritabanlarının listesini döndürür.

2. Hivesampletable tablosundaki verileri delays tablosuna aktarmak için aşağıdaki komutu kullanın:

   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 tablosunu içeren veritabanına bağlanır ve verileri dizinden /tutorials/flightdelays/outputdelays tablosuna aktarır.

3. sqoop Komut tamamlandıktan sonra veritabanına bağlanmak için tsql yardımcı programını kullanın:

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

4. Verilerin delays tablosuna aktarıldığını doğrulamak için aşağıdaki deyimleri kullanın:

   SELECT * FROM delays
   GO
   

   Tabloda verilerin listesini görürsünüz. Tablo, şehir adını ve bu şehre ait ortalama uçuş gecikme süresini içerir.

5. tsql yardımcı programından çıkmak için girin exit .

Kaynakları temizleme

Bu öğreticide kullanılan tüm kaynaklar önceden var. Temizlemeye gerek yoktur.

Sonraki adımlar

HDInsight'ta verilerle çalışmanın diğer yollarını öğrenmek için aşağıdaki makaleye bakın:

Azure HDInsight kümeleriyle Azure Data Lake Storage 2. Nesil hizmetini kullanma