Eğitim Kılavuzu: Lakeflow Pipelines Düzenleyicisi'ni kullanarak ilk işlem hattınızı oluşturun

Otomatik Yükleyici ile veri düzenlemesi için yeni bir Lakeflow işlem hattı oluşturursunuz, ardından verileri temizleyip ilk 100 kullanıcıyı bulmak için bir sorgu oluşturarak örnek işlem hattını genişletirsiniz.

Bu öğreticide, Lakeflow Pipelines Düzenleyicisi'ni kullanarak şunları yapmayı öğreneceksiniz:

  • Varsayılan klasör yapısına sahip yeni bir işlem hattı oluşturun ve bir örnek dosya kümesiyle başlayın.
  • Beklentileri kullanarak veri kalitesi kısıtlamalarını tanımlayın.
  • Verileriniz üzerinde analiz gerçekleştirmek üzere işlem hattını yeni bir dönüşümle genişletmek için düzenleyici özelliklerini kullanın.

Gereksinimler

Bu öğreticiye başlamadan önce şunları yapmalısınız:

  • Azure Databricks çalışma alanında oturum açabilirsiniz.
  • Çalışma alanınız için Unity Kataloğu'nu etkinleştirin.
  • İşlem kaynağı oluşturma veya işlem kaynağına erişim iznine sahip olun.
  • Katalogda yeni şema oluşturma izinlerine sahip olun. Gereken izinler ALL PRIVILEGES veya USE CATALOG ve CREATE SCHEMA.
  • İşlem hatlarını ve bunların çıkışını oluşturmak, çalıştırmak, yenilemek ve görüntülemek için gereken tüm ayrıcalıklar için bkz. İşlem hatları için kimlikleri, izinleri ve ayrıcalıkları yönetme.

1. Adım: İşlem hattı oluşturma

Bu adımda, varsayılan klasör yapısını ve kod örneklerini kullanarak bir işlem hattı oluşturursunuz. Kod örnekleri, örnek veri kaynağındaki users tablosuna referans verirwanderbricks.

  1. Azure Databricks çalışma alanınızda Plus icon.Yeni, ardından Pipeline icon.ETL işlem hattı. Bu, New Pipeline <date> <time> gibi varsayılan bir işlem hattı adıyla işlem hattı düzenleyicisini açar.

  2. (İsteğe bağlı) Adı seçin ve işlem hattı için açıklayıcı bir ad girin.

  3. (İsteğe bağlı) Adın sağındaki kataloğa ve şemaya tıklayarak farklı varsayılanlar ayarlayın.

  4. (İsteğe bağlı) Sizin için oluşturulan my_transformation kaynak dosyasında, dil açılan listesinden Python veya SQL seçin.

  5. Kod simgesine tıklayın. Örnek kodu kullanın.

    Seçtiğiniz dildeki örnek kod, my_transformation klasöründeki transformations kaynak dosyasında yer alır. Çıktı veri kümeleri henüz oluşturulmamıştır ve ekranın sağ tarafındaki İşlem Hattı grafiği boştur.

  6. İşlem hattı kodunu (klasördeki transformations kod) çalıştırmak için ekranın sağ üst kısmındaki İşlem hattını çalıştır'a tıklayın.

    Çalıştırma işlemi tamamlandıktan sonra, çalışma alanının alt kısmında oluşturulan iki yeni tablo, sample_users_<date_time> ve sample_aggregation_<date_time>, gösterilir. Çalışma alanının sağ tarafındaki Pipeline grafiği artık iki tabloyu gösteriyor; buna sample_users öğesinin sample_aggregation için kaynak olduğu bilgisi de dahildir. Tam sample_users_<date_time> tablo adını not edin; sonraki adımda buna başvuracaksınız.

2. Adım: Veri kalitesi denetimleri uygulama

Bu adımda, tabloya sample_users bir veri kalitesi denetimi eklersiniz. Verileri kısıtlamak için işlem hattı beklentilerini kullanırsınız. Bu durumda, geçerli bir e-posta adresi olmayan tüm kullanıcı kayıtlarını siler ve temizlenen tablonun çıkışını olarak users_cleanedverirsiniz.

  1. Soldaki işlem hattı varlık tarayıcısındaArtı simgesine tıklayın ve Dönüştürme'yi seçin.

  2. Yeni dönüştürme dosyası oluştur iletişim kutusunda aşağıdaki seçimleri yapın:

    • Language için Python veya SQL seçin. Bunun önceki seçiminizle eşleşmesi gerekmez.
    • Dosyaya bir ad verin. Bu durumda users_cleaned seçin.
    • Hedef yolu için varsayılan değeri değiştirmeyin.
    • Veri kümesi türü için hiçbiri seçili olarak bırakın veya Gerçekleştirilmiş görünümü seçin. Gerçekleştirilmiş görünüm'ü seçerseniz, sizin için örnek kod oluşturur.

  3. Dönüştürme kodu dosyasını oluşturmak için Oluştur'a tıklayın.

  4. Yeni kod dosyanızda, kodu aşağıdakilerle eşleşecek şekilde düzenleyin (önceki ekrandaki seçiminize bağlı olarak SQL veya Python kullanın). sample_users_<date_time> ifadesini, önceki bölümdeki sample_users tablonuzun tam adıyla değiştirin.

    SQL

    -- Drop all rows that do not have an email address

CREATE MATERIALIZED VIEW users_cleaned
(
  CONSTRAINT non_null_email EXPECT (email IS NOT NULL) ON VIOLATION DROP ROW
) AS
SELECT *
FROM sample_users_<date_time>;

    Python

    from pyspark import pipelines as dp

# Drop all rows that do not have an email address

@dp.materialized_view
@dp.expect_or_drop("no null emails", "email IS NOT NULL")
def users_cleaned():
    return (
        spark.read.table("sample_users_<date_time>")
    )

  5. İşlem hattını güncelleştirmek için İşlem hattını çalıştır'a tıklayın. Artık üç tablosu olmalıdır.

3. Adım: En çok kullanılan kullanıcıları analiz etme

Ardından, oluşturdukları rezervasyon sayısına göre ilk 100 kullanıcıyı edinin. wanderbricks.bookings tabloyu users_cleaned maddileştirilmiş görünüme birleştirin.

  1. Soldaki işlem hattı varlık tarayıcısında Artı simgesine tıklayın ve Dönüştürme seçeneğini belirleyin.

  2. Yeni dönüştürme dosyası oluştur iletişim kutusunda aşağıdaki seçimleri yapın:

    • Language için Python veya SQL seçin. Bunun önceki seçimlerinizle eşleşmesi gerekmez.
    • Dosyaya bir ad verin. Bu durumda users_and_bookings seçin.
    • Hedef yolu için varsayılan değeri değiştirmeyin.
    • Veri kümesi türü için Seçili değil olarak bırakın.

  3. Dönüştürme kodu dosyasını oluşturmak için Oluştur'a tıklayın.

  4. Yeni kod dosyanızda, kodu aşağıdakilerle eşleşecek şekilde düzenleyin (önceki ekrandaki seçiminize bağlı olarak SQL veya Python kullanın).

    SQL

    -- Get the top 100 users by number of bookings

CREATE OR REFRESH MATERIALIZED VIEW users_and_bookings AS
SELECT u.name AS name, COUNT(b.booking_id) AS booking_count
FROM users_cleaned u
JOIN samples.wanderbricks.bookings b ON u.user_id = b.user_id
GROUP BY u.name
ORDER BY booking_count DESC
LIMIT 100;

    Python

    from pyspark import pipelines as dp
from pyspark.sql.functions import col, count, desc

# Get the top 100 users by number of bookings

@dp.materialized_view
def users_and_bookings():
    return (
        spark.read.table("users_cleaned")
        .join(spark.read.table("samples.wanderbricks.bookings"), "user_id")
        .groupBy(col("name"))
        .agg(count("booking_id").alias("booking_count"))
        .orderBy(desc("booking_count"))
        .limit(100)
    )

  5. Veri kümelerini güncelleştirmek için İşlem hattını çalıştır'a tıklayın. Çalıştırma tamamlandığında, yeni tablo da dahil olmak üzere dört tablo olduğunu users_and_bookings'nde görebilirsiniz.

    İşlem hattındaki dört tabloyu gösteren grafik

Ek kaynaklar

Lakeflow işlem hatları düzenleyicisinin bazı özelliklerini kullanmayı öğrendiğinize ve bir işlem hattı oluşturduğunuza göre, hakkında daha fazla bilgi edinmek için diğer bazı özellikler şunlardır:

  • Last updated on