Dateningenieure und Wissenschaftler arbeiten häufig mit einer Vielzahl von Tools, um verschiedene Arten von Datenvorgängen abzuwickeln – von der verteilten Verarbeitung in großem Maßstab bis hin zur In-Memory-Datenmanipulation. Das Docker-Image alexmerced/spark35nb vereinfacht dies, indem es eine vorkonfigurierte Umgebung bietet, in der Sie mit mehreren beliebten Datentools experimentieren können, darunter PySpark, Pandas, DuckDB, Polars und DataFusion.
In diesem Blog führen wir Sie durch die Einrichtung dieser Umgebung und zeigen, wie Sie mit diesen Tools grundlegende Datenvorgänge wie das Schreiben von Daten, das Laden von Daten und das Ausführen von Abfragen und Aggregationen durchführen. Ganz gleich, ob Sie mit großen Datensätzen arbeiten oder nur kleine In-Memory-Daten bearbeiten müssen, Sie werden sehen, wie sich diese verschiedenen Bibliotheken gegenseitig ergänzen können.
Um zu beginnen, müssen Sie zunächst das Docker-Image alexmerced/spark35nb vom Docker Hub abrufen. Dieses Image wird mit einer vorkonfigurierten Umgebung geliefert, die Spark 3.5.2, JupyterLab und viele beliebte Datenbearbeitungsbibliotheken wie Pandas, DuckDB und Polars umfasst.
Führen Sie den folgenden Befehl aus, um das Bild abzurufen:
docker pull alexmerced/spark35nb
Als nächstes führen Sie den Container mit dem folgenden Befehl aus:
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
Sobald der Container betriebsbereit ist, öffnen Sie Ihren Browser und navigieren Sie zu localhost:8888, um auf JupyterLab zuzugreifen, wo Sie alle Ihre Datenoperationen durchführen.
Da Sie nun Ihre Umgebung eingerichtet haben, können wir mit der Durchführung einiger grundlegender Datenoperationen mit PySpark, Pandas, DuckDB, Polars und DataFusion fortfahren.
PySpark ist die Python-API für Apache Spark, eine Open-Source-Engine, die für die Verarbeitung großer Datenmengen und verteiltes Rechnen entwickelt wurde. Sie können mit Big Data arbeiten, indem Sie Daten und Berechnungen über einen Cluster verteilen. Während Spark normalerweise in einem verteilten Cluster ausgeführt wird, können Sie es mit diesem Setup lokal auf einem einzelnen Knoten ausführen – perfekt für Entwicklung und Tests.
Mit PySpark können Sie Datenmanipulationen, SQL-Abfragen, maschinelles Lernen und mehr durchführen – und das alles in einem Framework, das Big Data effizient verarbeitet. In diesem Abschnitt erfahren Sie, wie Sie Daten mit PySpark in der JupyterLab-Umgebung schreiben und abfragen.
Beginnen wir mit der Erstellung eines einfachen Datensatzes in PySpark. Initialisieren Sie zunächst eine Spark-Sitzung, die für die Interaktion mit der Funktionalität von Spark erforderlich ist. Wir erstellen einen kleinen DataFrame mit Beispieldaten und zeigen ihn an.
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()
In diesem Beispiel haben wir einen DataFrame mit drei Datenzeilen erstellt, die die Namen und das Alter von Personen darstellen. Mit der Funktion df.show() können wir den Inhalt des DataFrame anzeigen und so die gerade erstellten Daten einfacher überprüfen.
Als nächstes laden wir einen Datensatz aus einer Datei und führen einige grundlegende Abfragen aus. PySpark kann verschiedene Dateiformate verarbeiten, darunter CSV, JSON und Parquet.
Nehmen wir für dieses Beispiel an, dass wir eine CSV-Datei mit weiteren Daten über Personen haben, die wir in einen DataFrame laden. Anschließend demonstrieren wir eine einfache Filterabfrage und Aggregation, um die Anzahl der Personen in jeder Altersgruppe zu zählen.
# 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()
In diesem Beispiel haben wir mit spark.read.csv() eine CSV-Datei in einen PySpark DataFrame geladen. Dann haben wir zwei verschiedene Operationen angewendet:
Mit PySpark können Sie komplexere Abfragen und Aggregationen für große Datensätze durchführen und es so zu einem Tool für die Verarbeitung großer Datenmengen machen.
Im nächsten Abschnitt beschäftigen wir uns mit Pandas, das sich hervorragend für kleinere In-Memory-Datenvorgänge eignet, die keine verteilte Verarbeitung erfordern.
Pandas ist eine der am häufigsten verwendeten Python-Bibliotheken zur Datenmanipulation und -analyse. Es bietet benutzerfreundliche Datenstrukturen wie DataFrames, mit denen Sie auf intuitive Weise mit Tabellendaten arbeiten können. Im Gegensatz zu PySpark, das für die verteilte Datenverarbeitung in großem Maßstab konzipiert ist, arbeitet Pandas im Arbeitsspeicher und ist daher ideal für kleine bis mittelgroße Datensätze.
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.
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.
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.
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.
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.
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.
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.
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.
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.
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.
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.
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.
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.
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.
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:
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
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.
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:
# 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)
# Query Dremio and return data as a Pandas DataFrame
df = dremio.toPandas("SELECT * FROM my_table;")
print(df)
# Query Dremio and return data as a Polars DataFrame
df_polars = dremio.toPolars("SELECT * FROM my_table;")
print(df_polars)
# 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.
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.
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.
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.
Das obige ist der detaillierte Inhalt vonErkunden von Datenoperationen mit PySpark, Pandas, DuckDB, Polars und DataFusion in einem Python-Notizbuch. Für weitere Informationen folgen Sie bitte anderen verwandten Artikeln auf der PHP chinesischen Website!
![[Web-Frontend] Node.js-Schnellstart](https://img.php.cn/upload/course/000/000/067/662b5d34ba7c0227.png)
