Veri dosyalarıyla çalışmak için Spark kullanma

8 dakika

Not defterini ayarladıktan ve kümeye ekledikten sonra veri dosyalarını okumak ve işlemek için Spark'ı kullanabilirsiniz. Spark; CSV, JSON, Parquet, ORC, Avro ve Delta gibi çok çeşitli biçimleri destekler ve Databricks çalışma alanında, Azure Data Lake'te veya Blob Depolama'da veya diğer dış sistemlerde depolanan dosyalara erişmek için yerleşik bağlayıcılar sağlar.

İş akışı genellikle üç adımı izler:

Spark.read dosyasını doğru biçim ve yol ile kullanarak spark DataFrame'de bir dosyayı okuyun. SPARK, CSV veya JSON gibi ham metin biçimlerini okurken şemayı (sütun adları ve veri türleri) çıkarsayabilir, ancak bu bazen yavaş veya güvenilir değildir. Üretimde daha iyi bir uygulama, verilerin tutarlı ve verimli bir şekilde yüklenmesi için şemayı açıkça tanımlamaktır.
SQL veya DataFrame işlemlerini kullanarak DataFrame'i keşfedin ve dönüştürün (örneğin, satırları filtreleme, sütunları seçme, değerleri toplama).
Sonuçları seçilen biçimde depolama alanına geri yazın.

Spark'ta dosyalarla çalışma, küçük ve büyük veri kümelerinde tutarlı olacak şekilde tasarlanmıştır. Spark işi kümeye dağıttığından, küçük bir CSV dosyasını test etmek için kullanılan kod çok daha büyük veri kümelerinde de çalışır. Bu, hızlı incelemeden daha karmaşık veri işlemeye kadar ölçeği artırmayı kolaylaştırır.

Verileri bir veri çerçevesine yükleme

Şimdi verilerle çalışmak için veri çerçevesini nasıl kullanabileceğinizi görmek için varsayımsal bir örneği inceleyelim. Unity Kataloğu biriminde depolanan products.csv adlı virgülle ayrılmış bir metin dosyasında aşağıdaki verilere sahip olduğunuzu varsayalım:

ProductID,ProductName,Category,ListPrice
771,"Mountain-100 Silver, 38",Mountain Bikes,3399.9900
772,"Mountain-100 Silver, 42",Mountain Bikes,3399.9900
773,"Mountain-100 Silver, 44",Mountain Bikes,3399.9900
...

Spark not defterinde, verileri bir veri çerçevesine yüklemek ve ilk 10 satırı görüntülemek için aşağıdaki PySpark kodunu kullanabilirsiniz:

%pyspark
df = spark.read.load('/Volumes/my_catalog/my_schema/my_volume/products.csv',
    format='csv',
    header=True
)
display(df.limit(10))

%pyspark Başlangıçtaki satır, bir özel komut olarak adlandırılır ve Spark'a bu hücrede kullanılan dilin PySpark olduğunu bildirir. Ürün verileri örneği için eşdeğer Scala kodu aşağıda verilmişti:

%spark
val df = spark.read.format("csv").option("header", "true").load("/Volumes/my_catalog/my_schema/my_volume/products.csv")
display(df.limit(10))

Not

Azure Databricks dosyalara Unity Kataloğu birimleri aracılığıyla /Volumes/<catalog>/<schema>/<volume>/<path> biçimindeki yollar kullanılarak erişilir. Bu, dosya depolama ve erişim için önerilen desendir. DBFS (/data/...), eski bir yaklaşımdır ve artık önerilmez.

Scala'yı belirtmek için %spark özelliği kullanılır.

İpucu

Not Defteri arabirimindeki her hücre için kullanmak istediğiniz dili de seçebilirsiniz.

Daha önce gösterilen örneklerin her ikisi de aşağıdaki gibi bir çıkış oluşturur:

ProductID	ProductName	Kategori	Liste Fiyatı
771	Mountain-100 Gümüş, 38	Dağ Bisikletleri	3399.9900
772	Mountain-100 Gümüş, 42	Dağ Bisikletleri	3399.9900
773	Mountain-100 Gümüş, 44	Dağ Bisikletleri	3399.9900
...	...	...	...

Veri çerçevesi şeması belirtme

Önceki örnekte CSV dosyasının ilk satırı sütun adlarını içeriyordu ve Spark, içerdiği verilerden her sütunun veri türünü çıkarabiliyordu. Ayrıca veriler için açık bir şema belirtebilirsiniz. Bu şema, sütun adları veri dosyasına eklenmediğinde yararlı olabilir, örneğin şu CSV örneği:

771,"Mountain-100 Silver, 38",Mountain Bikes,3399.9900
772,"Mountain-100 Silver, 42",Mountain Bikes,3399.9900
773,"Mountain-100 Silver, 44",Mountain Bikes,3399.9900
...

Aşağıdaki PySpark örneğinde, product-data.csv adlı bir dosyadan yüklenecek veri çerçevesi için bu biçimde bir şemanın nasıl belirtileceği gösterilmektedir:

from pyspark.sql.types import *
from pyspark.sql.functions import *

productSchema = StructType([
    StructField("ProductID", IntegerType()),
    StructField("ProductName", StringType()),
    StructField("Category", StringType()),
    StructField("ListPrice", FloatType())
    ])

df = spark.read.load('/Volumes/my_catalog/my_schema/my_volume/product-data.csv',
    format='csv',
    schema=productSchema,
    header=False)
display(df.limit(10))

Sonuçlar bir kez daha şuna benzer olacaktır:

ProductID	ProductName	Kategori	Liste Fiyatı
771	Mountain-100 Gümüş, 38	Dağ Bisikletleri	3399.9900
772	Mountain-100 Gümüş, 42	Dağ Bisikletleri	3399.9900
773	Mountain-100 Gümüş, 44	Dağ Bisikletleri	3399.9900
...	...	...	...

Veri çerçevelerini filtreleme ve gruplandırma

Dataframe sınıfının yöntemlerini kullanarak içerdiği verileri filtreleyebilir, sıralayabilir, gruplandırabilir ve başka şekilde işleyebilirsiniz. Örneğin, aşağıdaki kod örneği önceki örnekteki select ürün verilerini içeren df veri çerçevesinden ProductName ve ListPrice sütunlarını almak için yöntemini kullanır:

pricelist_df = df.select("ProductID", "ListPrice")

Bu kod örneğinden alınan sonuçlar şuna benzer olacaktır:

ProductID	Liste fiyatı
771	3399.9900
772	3399.9900
773	3399.9900
...	...

Çoğu veri işleme yönteminde ortak olarak, select yeni bir veri çerçevesi nesnesi döndürür.

İpucu

Bir veri çerçevesinden sütunların alt kümesini seçmek yaygın bir işlemdir ve bu işlem aşağıdaki daha kısa söz dizimi kullanılarak da yapılabilir:

pricelist_df = df["ProductID", "ListPrice"]

Dönüştürülen bir veri çerçevesiyle sonuçlanan bir dizi işleme gerçekleştirmek için yöntemleri birlikte "zincirleyebilirsiniz". Örneğin, bu örnek kodselect veya where kategorisine sahip ürünler için ProductName ve ListPrice sütunlarını içeren yeni bir veri çerçevesi oluşturmak için ve yöntemlerini zincirler:

bikes_df = df.select("ProductName", "ListPrice").where((df["Category"]=="Mountain Bikes") | (df["Category"]=="Road Bikes"))
display(bikes_df)

Bu kod örneğinden alınan sonuçlar şuna benzer olacaktır:

ProductName	Liste Fiyatı
Mountain-100 Gümüş, 38	3399.9900
Road-750 Siyah, 52	539.9900
...	...

Verileri gruplandırmak ve toplamak için yöntemini ve toplama işlevlerini kullanabilirsiniz groupby . Örneğin, aşağıdaki PySpark kodu her kategori için ürün sayısını sayar:

counts_df = df.select("ProductID", "Category").groupBy("Category").count()
display(counts_df)

Bu kod örneğinden alınan sonuçlar şuna benzer olacaktır:

Kategori	sayı
Kulaklık	3
Tekerlekler	14
Dağ Bisikletleri	32
...	...

Not

Spark DataFrame'ler deklaratif ve değişmezdir. Her dönüşüm (örneğin select, filter veya groupBy gibi), nasıl çalıştığını değil, istediğinizi temsil eden yeni bir DataFrame oluşturur. Bu, kodu yeniden kullanılabilir, iyileştirilebilir ve yan etkilerden arındırılır hale getirir. Ancak bu dönüştürmelerin hiçbiri, siz bir eylem (örneğin, display, collect, write) tetikleyene kadar gerçekleşmez; bu noktada Spark, tamamen optimize edilmiş planı çalıştırır.

Spark'ta SQL ifadelerini kullanma

Veri Çerçevesi API'si, veri analistlerinin verileri sorgulamak ve işlemek için SQL ifadelerini kullanmasına olanak tanıyan Spark SQL adlı spark kitaplığının bir parçasıdır.

Spark kataloğunda veritabanı nesneleri oluşturma

Spark kataloğu, görünümler ve tablolar gibi ilişkisel veri nesneleri için bir meta veri deposudur. Spark çalışma zamanı, spark destekli herhangi bir dilde yazılmış kodu bazı veri analistleri veya geliştiriciler için daha doğal olabilecek SQL ifadeleriyle sorunsuz bir şekilde tümleştirmek için kataloğu kullanabilir.

Aşağıdaki kod örneğinde gösterildiği gibi, bir veri çerçevesindeki verileri Spark kataloğunda sorgulama için kullanılabilir hale getirmenin en basit yollarından biri geçici bir görünüm oluşturmaktır:

df.createOrReplaceTempView("products")

Görünüm geçicidir, yani geçerli oturumun sonunda otomatik olarak silinir. Spark SQL kullanılarak sorgulanabilen bir veritabanı tanımlamak için katalogda kalıcı olan tablolar da oluşturabilirsiniz.

Not

Bu modülde Spark katalog tablolarını derinlemesine incelemeyeceğiz, ancak birkaç önemli noktayı vurgulamak için zaman ayırabilirsiniz:

yöntemini kullanarak spark.catalog.createTable boş bir tablo oluşturabilirsiniz. Tablolar, temel verilerini katalogla ilişkili depolama konumunda depolayan meta veri yapılarıdır. Bir tablonun silinmesi, temel alınan verileri de siler.
Yöntemini kullanarak bir veri çerçevesini saveAsTable tablo olarak kaydedebilirsiniz.
spark.catalog.createExternalTable yöntemini kullanarak bir harici tablo oluşturabilirsiniz. Dış tablolar katalogda meta verileri tanımlar ancak temel verilerini bir dış depolama konumundan alır; genellikle bir veri gölündeki bir klasör. Dış tablo silindiğinde temel alınan veriler silinmez.

Verileri sorgulamak için Spark SQL API'sini kullanma

Spark SQL API'sini herhangi bir dilde yazılmış kodda kullanarak katalogdaki verileri sorgulayabilirsiniz. Örneğin, aşağıdaki PySpark kodu , ürünler görünümünden veri çerçevesi olarak veri döndürmek için bir SQL sorgusu kullanır.

bikes_df = spark.sql("SELECT ProductID, ProductName, ListPrice \
                      FROM products \
                      WHERE Category IN ('Mountain Bikes', 'Road Bikes')")
display(bikes_df)

Kod örneğinden elde edilen sonuçlar aşağıdaki tabloya benzer olacaktır:

ProductName	Liste Fiyatı
Mountain-100 Gümüş, 38	3399.9900
Road-750 Siyah, 52	539.9900
...	...

SQL kodunu kullanma

Önceki örnekte, Spark koduna SQL ifadelerini eklemek için Spark SQL API'sinin nasıl kullanılacağı gösterilmiştir. Not defterinde, aşağıdaki gibi katalogdaki nesneleri sorgulayan SQL kodunu çalıştırmak için de büyü kullanabilirsiniz %sql :

%sql

SELECT Category, COUNT(ProductID) AS ProductCount
FROM products
GROUP BY Category
ORDER BY Category

SQL kodu örneği, not defterinde aşağıdaki gibi otomatik olarak tablo olarak görüntülenen bir sonuç kümesi döndürür:

Kategori	Ürün Sayısı
Bib-Şort	3
Bisiklet Rafları	1
Bisiklet Standları	1
...	...

Geri Bildirim

Bu sayfayı yararlı buldunuz mu?