Meneroka Operasi Data dengan PySpark, Pandas, DuckDB, Polars dan DataFusion dalam Buku Nota Python

• Kursus Kemalangan Apache Iceberg: Apakah Data Lakehouse dan Format Jadual?
• Salinan Percuma Apache Iceberg Panduan Definitif
• Kursus Perlumbaan Apache Iceberg Percuma
• Senarai Main Video Kejuruteraan Iceberg Lakehouse

Jurutera dan saintis data sering bekerja dengan pelbagai alatan untuk mengendalikan jenis operasi data yang berbeza—daripada pemprosesan teragih berskala besar kepada manipulasi data dalam memori. Imej Docker alexmerced/spark35nb memudahkan perkara ini dengan menawarkan persekitaran pra-konfigurasi yang membolehkan anda mencuba berbilang alatan data popular, termasuk PySpark, Pandas, DuckDB, Polars dan DataFusion.

Dalam blog ini, kami akan membimbing anda melalui penyediaan persekitaran ini dan menunjukkan cara melaksanakan operasi data asas seperti menulis data, memuatkan data dan melaksanakan pertanyaan dan pengagregatan menggunakan alatan ini. Sama ada anda berurusan dengan set data yang besar atau hanya perlu memanipulasi data dalam memori yang kecil, anda akan melihat bagaimana perpustakaan berbeza ini boleh saling melengkapi.

Bahagian 1: Sediakan Persekitaran Anda

1.1 Tarik Imej Docker

Untuk bermula, anda perlu menarik imej Docker alexmerced/spark35nb daripada Docker Hub. Imej ini disertakan dengan persekitaran prakonfigurasi yang termasuk Spark 3.5.2, JupyterLab dan banyak perpustakaan manipulasi data popular seperti Pandas, DuckDB dan Polars.

Jalankan arahan berikut untuk menarik imej:

docker pull alexmerced/spark35nb

Seterusnya, jalankan bekas menggunakan arahan berikut:

docker run -p 8888:8888 -p 4040:4040 -p 7077:7077 -p 8080:8080 -p 18080:18080 -p 6066:6066 -p 7078:7078 -p 8081:8081  alexmerced/spark35nb

Setelah bekas siap dan berjalan, buka penyemak imbas anda dan navigasi ke localhost:8888 untuk mengakses JupyterLab, tempat anda akan melaksanakan semua operasi data anda.

Sekarang anda telah menyediakan persekitaran anda, kami boleh meneruskan untuk melaksanakan beberapa operasi data asas menggunakan PySpark, Pandas, DuckDB, Polars dan DataFusion.

Bahagian 2: Bekerja dengan PySpark

2.1 Apakah PySpark?

PySpark ialah API Python untuk Apache Spark, enjin sumber terbuka yang direka untuk pemprosesan data berskala besar dan pengkomputeran teragih. Ia membolehkan anda bekerja dengan data besar dengan mengedarkan data dan pengiraan merentas kluster. Walaupun Spark biasanya dijalankan dalam kelompok teragih, persediaan ini membolehkan anda menjalankannya secara setempat pada satu nod—sempurna untuk pembangunan dan ujian.

Menggunakan PySpark, anda boleh melakukan manipulasi data, pertanyaan SQL, pembelajaran mesin dan banyak lagi, semuanya dalam rangka kerja yang mengendalikan data besar dengan cekap. Dalam bahagian ini, kami akan membincangkan cara menulis dan membuat pertanyaan data menggunakan PySpark dalam persekitaran JupyterLab.

2.2 Menulis Data dengan PySpark

Mari mulakan dengan mencipta set data ringkas dalam PySpark. Mula-mula, mulakan sesi Spark, yang diperlukan untuk berinteraksi dengan fungsi Spark. Kami akan mencipta DataFrame kecil dengan data sampel dan memaparkannya.

from pyspark.sql import SparkSession

# Initialize the Spark session
spark = SparkSession.builder.appName("PySpark Example").getOrCreate()

# Sample data: a list of tuples containing names and ages
data = [("Alice", 34), ("Bob", 45), ("Catherine", 29)]

# Create a DataFrame
df = spark.createDataFrame(data, ["Name", "Age"])

# Show the DataFrame
df.show()

Dalam contoh ini, kami mencipta DataFrame dengan tiga baris data, yang mewakili nama dan umur orang. Fungsi df.show() membolehkan kami memaparkan kandungan DataFrame, menjadikannya mudah untuk memeriksa data yang baru kami buat.

2.3 Memuatkan dan Menyoal Data dengan PySpark

Seterusnya, mari muatkan set data daripada fail dan jalankan beberapa pertanyaan asas. PySpark boleh mengendalikan pelbagai format fail, termasuk CSV, JSON dan Parket.

Untuk contoh ini, mari kita anggap kita mempunyai fail CSV dengan lebih banyak data tentang orang, yang akan kita muatkan ke dalam DataFrame. Kemudian kami akan menunjukkan pertanyaan penapis dan pengagregatan mudah untuk mengira bilangan orang dalam setiap kumpulan umur.

# Load a CSV file into a DataFrame
df_csv = spark.read.csv("data/people.csv", header=True, inferSchema=True)

# Show the first few rows of the DataFrame
df_csv.show()

# Filter the data to only include people older than 30
df_filtered = df_csv.filter(df_csv["Age"] > 30)

# Show the filtered DataFrame
df_filtered.show()

# Group by Age and count the number of people in each age group
df_grouped = df_csv.groupBy("Age").count()

# Show the result of the grouping
df_grouped.show()

Dalam contoh ini, kami memuatkan fail CSV ke dalam PySpark DataFrame menggunakan spark.read.csv(). Kemudian, kami menggunakan dua operasi berbeza:

• Penapisan: Kami menapis DataFrame untuk menunjukkan hanya baris yang umurnya melebihi 30.
• Pengagregatan: Kami mengumpulkan data mengikut umur dan mengira bilangan orang dalam setiap kumpulan umur.

Dengan PySpark, anda boleh melakukan pertanyaan dan pengagregatan yang lebih kompleks pada set data yang besar, menjadikannya alat untuk pemprosesan data besar.

Dalam bahagian seterusnya, kami akan meneroka Panda, yang bagus untuk operasi data dalam memori yang lebih kecil yang tidak memerlukan pemprosesan teragih.

Bahagian 3: Manipulasi Data dengan Panda

3.1 Apakah Panda?

Panda ialah salah satu perpustakaan Python yang paling banyak digunakan untuk manipulasi dan analisis data. Ia menyediakan struktur data yang mudah digunakan, seperti DataFrames, yang membolehkan anda bekerja dengan data jadual dengan cara yang intuitif. Tidak seperti PySpark, yang direka untuk pemprosesan data teragih berskala besar, Pandas berfungsi dalam ingatan, menjadikannya sesuai untuk set data bersaiz kecil hingga sederhana.

With Pandas, you can read and write data from various formats, including CSV, Excel, and JSON, and perform common data operations like filtering, aggregating, and merging data with simple and readable syntax.

3.2 Loading Data with Pandas

Let’s start by loading a dataset into a Pandas DataFrame. We’ll read a CSV file, which is a common file format for data storage, and display the first few rows.

import pandas as pd

# Load a CSV file into a Pandas DataFrame
df_pandas = pd.read_csv("data/people.csv")

# Display the first few rows of the DataFrame
print(df_pandas.head())

In this example, we read the CSV file people.csv using pd.read_csv() and loaded it into a Pandas DataFrame. The head() method lets you view the first few rows of the DataFrame, which is useful for quickly inspecting the data.

3.3 Basic Operations with Pandas

Now that we have loaded the data, let’s perform some basic operations, such as filtering rows and grouping data. Pandas allows you to apply these operations easily with simple Python syntax.

# Filter the data to show only people older than 30
df_filtered = df_pandas[df_pandas["Age"] > 30]

# Display the filtered data
print(df_filtered)

# Group the data by 'Age' and count the number of people in each age group
df_grouped = df_pandas.groupby("Age").count()

# Display the grouped data
print(df_grouped)

Here, we filtered the data to include only people older than 30 using a simple boolean expression. Then, we used the groupby() function to group the DataFrame by age and count the number of people in each age group.

Pandas is incredibly efficient for in-memory data operations, making it a go-to tool for smaller datasets that can fit in your machine's memory. In the next section, we’ll explore DuckDB, a SQL-based tool that enables fast querying over in-memory data.

Section 4: Exploring DuckDB

4.1 What is DuckDB?

DuckDB is an in-memory SQL database management system (DBMS) designed for analytical workloads. It offers high-performance, efficient querying of datasets directly within your Python environment. DuckDB is particularly well-suited for performing complex SQL queries on structured data, like CSVs or Parquet files, without needing to set up a separate database server.

DuckDB is lightweight, yet powerful, and can be used as an alternative to tools like SQLite, especially when working with analytical queries on large datasets.

4.2 Writing Data into DuckDB

DuckDB can easily integrate with Pandas, allowing you to transfer data from a Pandas DataFrame into DuckDB for SQL-based queries. Here’s how to create a table in DuckDB using the data from Pandas.

import duckdb

# Connect to an in-memory DuckDB instance
conn = duckdb.connect()

# Create a table in DuckDB from the Pandas DataFrame
conn.execute("CREATE TABLE people AS SELECT * FROM df_pandas")

# Show the content of the 'people' table
conn.execute("SELECT * FROM people").df()

In this example, we connected to DuckDB and created a new table people from the Pandas DataFrame df_pandas. DuckDB’s execute() function allows you to run SQL commands, making it easy to interact with data using SQL queries.

4.3 Querying Data in DuckDB

Once your data is loaded into DuckDB, you can run SQL queries to filter, aggregate, and analyze your data. DuckDB supports a wide range of SQL functionality, making it ideal for users who prefer SQL over Python for data manipulation.

# Query to select people older than 30
result = conn.execute("SELECT Name, Age FROM people WHERE Age > 30").df()

# Display the result of the query
print(result)

# Query to group people by age and count the number of people in each age group
result_grouped = conn.execute("SELECT Age, COUNT(*) as count FROM people GROUP BY Age").df()

# Display the grouped result
print(result_grouped)

In this example, we used SQL to filter the people table, selecting only those who are older than 30. We then ran a grouping query to count the number of people in each age group.

DuckDB is an excellent choice when you need SQL-like functionality directly in your Python environment. It allows you to leverage the power of SQL without the overhead of setting up and managing a database server. In the next section, we will explore Polars, a DataFrame library known for its speed and efficiency.

Section 5: Leveraging Polars for Fast DataFrame Operations

5.1 What is Polars?

Polars is a DataFrame library designed for high-performance data manipulation. It’s known for its speed and efficiency, particularly when compared to libraries like Pandas. Polars is written in Rust and uses an optimized query engine to handle large datasets quickly and with minimal memory usage. It also provides a similar interface to Pandas, making it easy to learn and integrate into existing Python workflows.

Polars is particularly well-suited for processing large datasets that might not fit into memory as easily or for scenarios where performance is a critical factor.

5.2 Working with Polars

Let’s start by creating a Polars DataFrame from a Python dictionary. We’ll then perform some basic operations like filtering and aggregating data.

import polars as pl

# Create a Polars DataFrame
df_polars = pl.DataFrame({
    "Name": ["Alice", "Bob", "Catherine"],
    "Age": [34, 45, 29]
})

# Display the Polars DataFrame
print(df_polars)

In this example, we created a Polars DataFrame using a Python dictionary. The syntax is similar to Pandas, but the operations are optimized for speed. Polars offers lazy evaluation, which means it can optimize the execution of multiple operations at once, reducing computation time.

5.3 Filtering and Aggregating with Polars

Now, let’s perform some common data operations such as filtering and aggregating the data. These operations are highly optimized in Polars and can be done using a simple and expressive syntax.

# Filter the DataFrame to show only people older than 30
df_filtered = df_polars.filter(pl.col("Age") > 30)

# Display the filtered DataFrame
print(df_filtered)

# Group by 'Age' and count the number of people in each age group
df_grouped = df_polars.groupby("Age").count()

# Display the grouped result
print(df_grouped)

In this example, we filtered the data to show only rows where the age is greater than 30, and then we grouped the data by age to count how many people are in each group. These operations are highly efficient in Polars due to its optimized memory management and query execution engine.

Polars is ideal when you need the speed of a DataFrame library for both small and large datasets, and when performance is a key requirement. Next, we will explore DataFusion, a tool for SQL-based querying over Apache Arrow data.

Section 6: DataFusion for Query Execution

6.1 What is DataFusion?

DataFusion is an in-memory query execution engine built on top of Apache Arrow, an efficient columnar memory format for analytics. It provides a powerful SQL engine that allows users to run complex queries over structured data stored in Arrow format. DataFusion is part of the Apache Arrow ecosystem, which aims to provide fast data interoperability across different data processing tools.

DataFusion is particularly well-suited for scenarios where you need to query large in-memory datasets using SQL without the overhead of traditional databases. Its integration with Arrow ensures that the data processing is both fast and memory-efficient.

6.2 Writing and Querying Data with DataFusion

DataFusion allows you to execute SQL queries on in-memory data using Apache Arrow. Let’s first create a DataFrame using DataFusion and then perform a few SQL queries on it.

from datafusion import SessionContext

# Initialize a DataFusion session
ctx = SessionContext()

# Create a DataFrame with some data
data = [
    {"Name": "Alice", "Age": 34},
    {"Name": "Bob", "Age": 45},
    {"Name": "Catherine", "Age": 29}
]

# Register the DataFrame as a table
df = ctx.create_dataframe(data)
ctx.register_table("people", df)

# Query the data to select people older than 30
result = ctx.sql("SELECT Name, Age FROM people WHERE Age > 30").collect()

# Display the result
print(result)

In this example, we used DataFusion’s SessionContext to create a DataFrame and registered it as a table. We then performed a simple SQL query to filter the data for people older than 30. DataFusion allows you to combine the power of SQL with the speed and efficiency of Apache Arrow’s in-memory format.

6.3 Aggregating Data with DataFusion

Just like in DuckDB, we can perform aggregation queries to group data by a specific field and count the number of records in each group. Let’s see how this works in DataFusion.

# Group by 'Age' and count the number of people in each age group
result_grouped = ctx.sql("SELECT Age, COUNT(*) as count FROM people GROUP BY Age").collect()

# Display the grouped result
print(result_grouped)

In this query, we grouped the data by the 'Age' column and counted how many people were in each age group. DataFusion’s SQL execution engine ensures that queries run efficiently, even on large datasets stored in-memory.

DataFusion is a great tool for users who need fast, SQL-based querying of large in-memory datasets and want to take advantage of Apache Arrow’s high-performance columnar data format. It’s particularly useful for building analytical pipelines that involve heavy querying of structured data.

Bonus Section: Integrating Dremio with Python

What is Dremio?

Dremio is a powerful data lakehouse platform that helps organizations unify and query their data from various sources. It enables users to easily govern, join, and accelerate queries on their data without the need for expensive and complex data warehouse infrastructures. Dremio's ability to access and query data directly from formats like Apache Iceberg, Delta Lake, S3, RDBMS, and JSON files, along with its performance enhancements, reduces the workload on traditional data warehouses.

Dremio is built on top of Apache Arrow, a high-performance columnar in-memory format, and utilizes Arrow Flight to accelerate the transmission of large datasets over the network. This integration provides blazing-fast query performance while enabling interoperability between various analytics tools.

In this section, we will demonstrate how to set up Dremio in a Docker container and use Python to query Dremio's data sources using the dremio-simple-query library.

6.1 Setting Up Dremio with Docker

To run Dremio on your local machine, use the following Docker command:

docker run -p 9047:9047 -p 31010:31010 -p 45678:45678 -p 32010:32010 -e DREMIO_JAVA_SERVER_EXTRA_OPTS=-Dpaths.dist=file:///opt/dremio/data/dist --name try-dremio dremio/dremio-oss

Once Dremio is up and running, navigate to http://localhost:9047 in your browser to access the Dremio UI. Here, you can configure your data sources, create virtual datasets, and explore the platform's capabilities.

6.2 Querying Dremio with Python using dremio-simple-query

The dremio-simple-query library allows you to query Dremio using Apache Arrow Flight, providing a high-performance interface for fetching and analyzing data from Dremio sources. With this library, you can easily convert Dremio queries into Pandas, Polars, or DuckDB DataFrames, or work directly with Apache Arrow data.

Here’s how to get started:

Step 1: Install the necessary libraries

Make sure you have the dremio-simple-query library installed (It is pre-installed on the alexmerced/spark35nb image). You can install it using pip:

pip install dremio-simple-query

Step 2: Set up your connection to Dremio

You’ll need your Dremio credentials to retrieve a token and establish a connection. Here’s a basic example:

from dremio_simple_query.connect import get_token, DremioConnection
from os import getenv
from dotenv import load_dotenv

# Load environment variables (TOKEN and ARROW_ENDPOINT)
load_dotenv()

# Login to Dremio and get a token
login_endpoint = "http://{host}:9047/apiv2/login"
payload = {
    "userName": "your_username",
    "password": "your_password"
}
token = get_token(uri=login_endpoint, payload=payload)

# Dremio Arrow Flight endpoint, make sure to put in the right host for your Dremio instance

arrow_endpoint = "grpc://{host}:32010"

# Establish connection to Dremio using Arrow Flight
dremio = DremioConnection(token, arrow_endpoint)

If you are running this locally using the docker run command, the host should be the IP address of the Dremio container on the docker network which you can find by running docker inspect.

In this code, we use the get_token function to retrieve an authentication token from Dremio's REST API and establish a connection to Dremio's Arrow Flight endpoint.

Step 3: Query Dremio and retrieve data in various formats

Once connected, you can use the connection to query Dremio and retrieve results in different formats, including Arrow, Pandas, Polars, and DuckDB. Here’s how:

Querying Data and Returning as Arrow Table:

# Query Dremio and return data as an Apache Arrow Table
stream = dremio.toArrow("SELECT * FROM my_table;")
arrow_table = stream.read_all()

# Display Arrow Table
print(arrow_table)

Converting to a Pandas DataFrame:

# Query Dremio and return data as a Pandas DataFrame
df = dremio.toPandas("SELECT * FROM my_table;")
print(df)

Converting to a Polars DataFrame:

# Query Dremio and return data as a Polars DataFrame
df_polars = dremio.toPolars("SELECT * FROM my_table;")
print(df_polars)

Querying with DuckDB:

# Query Dremio and return as a DuckDB relation
duck_rel = dremio.toDuckDB("SELECT * FROM my_table")

# Perform a query on the DuckDB relation
result = duck_rel.query("my_table", "SELECT * FROM my_table WHERE Age > 30").fetchall()

# Display results
print(result)

With the dremio-simple-query library, you can efficiently query large datasets from Dremio and immediately start analyzing them with various tools like Pandas, Polars, and DuckDB, all while leveraging the high-performance Apache Arrow format under the hood.

6.3 Why Use Dremio?

Dremio provides several benefits that make it a powerful addition to your data stack:

• Governance: Centralize governance over all your data sources, ensuring compliance and control.

• Data Federation: Join data across various sources, such as Iceberg, Delta Lake, JSON, CSV, and relational databases, without moving the data.

• Performance: Accelerate your queries with the help of Dremio's query acceleration features and Apache Arrow Flight.

• Cost Savings: By offloading workloads from traditional data warehouses, Dremio can reduce infrastructure costs.

Dremio's close relationship with Apache Arrow ensures that your queries are both fast and efficient, allowing you to seamlessly integrate various data sources and tools into your analytics workflows.

Conclusion

In this blog, we explored how to use a variety of powerful tools for data operations within a Python notebook environment. Starting with the alexmerced/spark35nb Docker image, we demonstrated how to set up a development environment that includes PySpark, Pandas, DuckDB, Polars, and DataFusion—each optimized for different data processing needs. We showcased basic operations like writing, querying, and aggregating data using each tool’s unique strengths.

• PySpark enables scalable, distributed processing for large datasets, perfect for big data environments.
• Pandas offers in-memory, easy-to-use data manipulation for smaller datasets, making it the go-to tool for quick data exploration.
• DuckDB provides an efficient, in-memory SQL engine, ideal for analytical queries without the need for complex infrastructure.
• Polars brings lightning-fast DataFrame operations, combining performance and simplicity for larger or performance-critical datasets.
• DataFusion, with its foundation in Apache Arrow, allows for high-performance SQL querying, particularly for analytical workloads in memory.

Finally, we introduced Dremio, which integrates with Apache Arrow to enable lightning-fast queries across a range of data sources. With the dremio-simple-query library, Dremio allows analysts to quickly fetch and analyze data using tools like Pandas, Polars, and DuckDB, ensuring that data is available when and where it's needed without the overhead of traditional data warehouses.

Whether you're working with small datasets or handling massive amounts of data in distributed environments, this setup provides a versatile, efficient, and scalable platform for any data engineering or data science project. By leveraging these tools together, you can cover the full spectrum of data processing, from exploration to large-scale analytics, with minimal setup and maximum performance.

Atas ialah kandungan terperinci Meneroka Operasi Data dengan PySpark, Pandas, DuckDB, Polars dan DataFusion dalam Buku Nota Python. Untuk maklumat lanjut, sila ikut artikel berkaitan lain di laman web China PHP!