DEV Community: Mike Houngbadji

DuckDB 🦆 : Libérez le puissant moteur de Requêtes

Mike Houngbadji — Fri, 04 Apr 2025 03:03:25 +0000

Introduction

DuckDB est largement reconnu pour ses capacités de traitement analytique, mais sa véritable puissance réside dans son moteur de requête exceptionnel. Contrairement aux bases de données traditionnelles qui nécessitent une configuration et des processus ETL complexes, DuckDB permet aux utilisateurs d'interroger des données provenant de diverses sources de manière fluide. Cet article explore le moteur de requête de DuckDB, ses avantages et propose un guide pratique pour l'utiliser efficacement.

Qu'est-ce qui fait du moteur de requête de DuckDB un outil révolutionnaire ?

Contrairement à de nombreuses bases de données qui nécessitent l'importation de données avant l'interrogation, le moteur de requête de DuckDB est conçu pour fonctionner directement sur plusieurs formats de fichiers et bases de données externes . Voici ce qui le distingue :

Caractéristiques principales

Prend en charge plusieurs formats de fichiers : Faites des requête sur des fichiers CSV, Parquet, JSON et plus sans conversion manuelle.
Fonctionne avec les bases de données relationnelles : se connecte de manière transparente à PostgreSQL, MySQL et autres.
Lire les formats de données ouverts : prend en charge Iceberg et Delta Lake.
Interface SQL simple : aucune configuration complexe requise.
Compatibilité multi-language : fonctionne avec Python, R, Java, Node.js, Rust, etc.
Pris en charge par les outils de visualisation : Pris en charge comme source de données par les outils de visualisation tels qu'Apache-Superset, metabase, les outils basés sur JDBC (Tableau,...), etc.

Optimisations avancées dans DuckDB

DuckDB est un système de gestion de base de données (SGBD) relationnel (orienté table) prenant en charge le langage SQL (Structured Query Language) . Il est conçu pour répondre au besoin d'un système de base de données offrant un ensemble unique de compromis, notamment pour les cas d'utilisation analytiques. Ses principales caractéristiques sont sa simplicité et son fonctionnement intégré , inspiré de SQLite. Il ne nécessite aucune dépendance externe pour la compilation ou l'exécution et est compilé dans un fichier d'en-tête et d'implémentation unique (une « fusion »), simplifiant ainsi le déploiement. Contrairement aux bases de données client-serveur traditionnelles, DuckDB s'exécute entièrement intégré à un processus hôte , éliminant ainsi le besoin d'installation et de maintenance d'un logiciel serveur distinct. Cette intégration permet un transfert de données à haut débit et la possibilité de traiter des données externes, telles que les Pandas DataFrames en Python, sans copie.
Conçu pour exceller dans les charges de travail de requêtes analytiques (OLAP) caractérisées par des requêtes complexes et longues sur de grands ensembles de données, DuckDB utilise un moteur d'exécution de requêtes vectorisé en colonnes . Cette approche traite les données par lots (« vecteurs »), réduisant considérablement la charge par rapport aux systèmes traditionnels basés sur les lignes comme PostgreSQL, MySQL ou SQLite. DuckDB est également extensible , permettant aux utilisateurs de définir de nouveaux types de données, fonctions, formats de fichiers et syntaxe SQL grâce à un mécanisme d'extension. Notamment, la prise en charge de formats courants comme Parquet et JSON, ainsi que de protocoles comme HTTP(S) et s3, est implémentée sous forme d'extensions.
DuckDB est extrêmement portable , capable d'être compilé pour tous les principaux systèmes d'exploitation (Linux, macOS, Windows) et architectures CPU (x86, ARM), et peut même fonctionner dans les navigateurs web et sur les téléphones portables via DuckDB-Wasm. Il fournit des API pour divers langages de programmation, dont Java, C, C++, Go et Python. Malgré sa simplicité, DuckDB est riche en fonctionnalités , offrant une prise en charge étendue des requêtes SQL complexes, une vaste bibliothèque de fonctions, des fonctions de fenêtrage, des garanties transactionnelles (propriétés ACID) , des bases de données monofichier persistantes et des index secondaires. Il est également profondément intégré à Python et R pour une analyse de données interactive et efficace.

Pour plus de détails techniques, visitez la documentation officielle de DuckDB .

DuckDB vs Apache Spark vs Trino : comparaison rapide

Fonctionnalité	DuckDB	Apache Spark	Trino
Coûts d'installation	Gratuit, léger	Configuration du cluster requise	Nécessite une configuration distribuée
Coûts de fonctionnement	Minimal	Dépenses courantes du cluster	Coûts du cluster ou du cloud
Courbe d'apprentissage	Connaissances de base en SQL	Nécessite un apprentissage plus approfondi	Connaissances SQL modérées
Performance	Rapide pour les requêtes locales	S'adapte bien au Big Data	Optimisé pour SQL distribué
Cas d'utilisation	Analyses locales	Traitement distribué des données	Moteur de requête fédéré

DuckDB est idéal pour les requêtes analytiques locales, Spark est meilleur pour le calcul distribué à grande échelle et Trino excelle dans les requêtes fédérées sur plusieurs sources.

Pratique : Libérez DuckDB comme moteur de requête

Dans ce tutoriel, nous utiliserons DuckDB pour interroger trois sources de données différentes :

Base de données PostgreSQL
Base de données MySQL
Fichiers CSV
JSON à partir d'un serveur Web

1. Configuration de l'environnement

Nous utiliserons Docker Compose pour configurer l'infrastructure. Commencez par cloner le dépôt et démarrer les services :

git clone https://github.com/mikekenneth/bp_duck_as_query_engine
cd bp_duck_as_query_engine
make up  # Start services

Vérifiez les conteneurs en cours d'exécution avec :

docker ps

Ensuite, accédez à DuckDB CLI :

make duckdb

Pour faciliter la configuration, j'ai créé un duckdb_init.sql contenant les commandes SQL nécessaires pour se connecter à nos sources de données externes.

2. Connexion aux sources de données externes

a) Connexion à MinIO (stockage compatible s3)

Pour interagir avec les données stockées dans MinIO, nous devons activer l'extension HTTPS et configurer les paramètres de connexion :

-- Enable the HTTPS Extension to connect to s3/Minio
INSTALL https;
LOAD https;

-- Set Connection settings
SET s3_region='us-east-1';
SET s3_url_style='path';
SET s3_endpoint='minio:9000';
SET s3_access_key_id='minio_root' ;
SET s3_secret_access_key='minio_toor';
SET s3_use_ssl=false; -- Needed when running without SSL

Vérifions si les paramètres sont correctement appliqués.

b) Connexion à PostgreSQL

Pour interroger les données dans notre instance PostgreSQL, nous devons charger l'extension PostgreSQL, puis nous attacher à la base de données :

-- Load Extension
INSTALL postgres;
LOAD postgres;

-- Attach the PostgreSQL database
ATTACH 'dbname=postgres user=postgres password=postgres host=postgres port=5432' 
AS postgres_db (TYPE postgres, SCHEMA 'public');

Explication : La INSTALL postgres;commande prend désormais en charge l'interaction avec les bases de données PostgreSQL. ATTACHElle établit une connexion à la base de données PostgreSQL spécifiée ( dbname, user, password, host) et lui attribue l'alias postgres_db, rendant ainsi ses tables accessibles dans DuckDB. Nous spécifions également le type et le schéma de la base de données.

Nous pouvons maintenant créer une table dans PostgreSQL en lisant un fichier CSV à partir de notre instance MinIO :

-- Create the base table in Postgres from s3
create or replace table postgres_db.fct_trips as
(
    select *
    from read_csv("s3://duckdb-bucket/init_data/base_raw.csv")
);

c) Connexion à MySQL

-- Load Extension
INSTALL mysql;
LOAD mysql;

-- Attach the MYSQL database
ATTACH 'database=db user=user password=password host=mysql port=3306' 
AS mysql_db (TYPE mysql);

Explication : Cette INSTALL mysql;commande permet d'interagir avec les bases de données MySQL. ATTACHElle se connecte à la base de données MySQL à l'aide des informations d'identification fournies et lui attribue l'alias mysql_db. Le type de base de données est également spécifié.

Créons une table dans MySQL en lisant un fichier CSV depuis MinIO :

-- Create the base table in MySQL from s3
create or replace table mysql_db.dim_credit_card as
(
    select *
    from read_csv("s3://duckdb-bucket/init_data/base_raw.csv")
);

Confirmons que les deux bases de données sont correctement connectées :

d) Requête de données JSON à partir d'un serveur Web

Enfin, démontrons comment récupérer des données directement à partir d’un serveur Web servant un fichier JSON :

select distinct *
from read_json('http://nginx:80/companies_data.json')

Cette requête lit et traite directement les données JSON disponibles à l'URL spécifiée.

Voici le résultat de la requête :

3. Interrogation simultanée de plusieurs sources de données

Avec la connectivité établie avec MinIO et nos bases de données relationnelles, nous pouvons désormais exécuter une seule requête qui récupère et joint les données de toutes nos sources configurées :

---------------- Query multiple Data Sources Simultaneously ----------------
with 
  fct_trips as (
    select *
    from postgres_db.public.fct_trips
  ),
  dim_customer as (
    select distinct *
    from read_csv('s3://duckdb-bucket/init_data/customer.csv')
  ),
  dim_creditcard as (
    select distinct *
    from mysql_db.db.dim_credit_card
  ),
  dim_companies_data_web as (
    select distinct *
    from read_json('http://nginx:80/companies_data.json')
  )
select 
  trips."Trip ID",
  -- Customer info
  dcust.id as customer_id, dcust.name as customer_name,dcust.sex as customer_sex,
  dcust.address as customer_address, dcust.birth_date as customer_birth_date,
  -- Credit Card info
  dcard.credit_card_number as credit_card_number, dcard.expire_date as credit_card_expire_date,
  dcard.provider as credit_card_provider, dcard.owner_name as credit_card_owner_name,
  -- Company info
  dcompany.name as company_name, dcompany.address as company_address,
  dcompany.created_date as company_created_date, dcompany.num_of_employee as company_num_of_employee
from fct_trips trips
LEFT JOIN dim_customer dcust on trips.customer_id = dcust.id
LEFT JOIN dim_creditcard dcard on trips.credit_card_number = dcard.credit_card_number
LEFT JOIN dim_companies_data_web dcompany on trips.Company = dcompany.name;

Résultat:

4. Exportation des résultats de la requête

De plus, DuckDB permet d'exporter directement les résultats de requêtes vers un stockage externe, tel que S3/MinIO, dans différents formats de fichiers. Ici, nous allons exporter les données jointes au format Parquet :

COPY (
with 
  fct_trips as (
    select *
    from postgres_db.public.fct_trips
...
...
...
from fct_trips trips
LEFT JOIN dim_customer dcust on trips.customer_id = dcust.id
LEFT JOIN dim_creditcard dcard on trips.credit_card_number = dcard.credit_card_number
LEFT JOIN dim_companies_data_web dcompany on trips.Company = dcompany.name

) TO 's3://duckdb-bucket/query_result.parquet' (FORMAT parquet);

Nous pouvons ensuite vérifier le contenu du fichier Parquet exporté directement avec DuckDB

Nous pouvons ensuite valider le fichier exporté :

SELECT * FROM read_parquet('s3://duckdb-bucket/query_result.parquet');

Conclusion

DuckDB est un moteur de requête puissant et léger qui permet d'interroger facilement plusieurs sources. Que vous travailliez avec des fichiers CSV, des bases de données ou des données JSON web, son interface SQL simple et ses hautes performances en font un excellent choix pour l'analyse. À mesure que l'écosystème se développe, DuckDB continue d'ajouter de nouvelles fonctionnalités, ce qui en fait une alternative solide pour le traitement local des données.

Références

DuckDB 🦆: Unleashing the Powerhouse Query Engine Within

Mike Houngbadji — Fri, 04 Apr 2025 02:38:01 +0000

Introduction

DuckDB is widely recognized for its analytical processing capabilities, but its true power lies in its exceptional query engine. Unlike traditional databases that require extensive setup and ETL processes, DuckDB allows users to query data from various sources seamlessly. This article explores DuckDB’s query engine, its advantages, and a hands-on guide to using it effectively.

What Makes DuckDB's Query Engine a Game-Changer?

Unlike many databases that require data to be imported before querying, DuckDB’s query engine is designed to work directly on multiple file formats and external databases. Here’s what makes it stand out:

Key Features

Supports Multiple File Formats: Query CSV, Parquet, JSON, and more without manual conversion.
Works with Relational Databases: Connects seamlessly to PostgreSQL, MySQL, and others.
Reads Open Data Formats: Supports Iceberg and Delta Lake.
Simple SQL Interface: No complex setup required.
Multi-language Compatibility: Works with Python, R, Java, Node.js, Rust, and more.
Supported by Visualizations tools: Supported as data-source by visualization tools such as Apache-Superset, metabase, JDBC based tools (Tableau,...), etc.

Advanced Optimizations in DuckDB

DuckDB is a relational (table-oriented) Database Management System (DBMS) that supports the Structured Query Language (SQL). It is designed to address the need for a database system that offers a unique set of trade-offs, particularly for analytical use cases. Key characteristics of DuckDB include its simplicity and embedded operation, drawing inspiration from SQLite. It has no external dependencies for compilation or runtime and is compiled into a single header and implementation file (an "amalgamation"), simplifying deployment. Unlike traditional client-server databases, DuckDB runs completely embedded within a host process, eliminating the need for separate server software installation and maintenance. This embedded nature allows for high-speed data transfer and the capability to process foreign data, such as Pandas DataFrames in Python, without copying.
Designed to excel in analytical query workloads (OLAP) characterized by complex, long-running queries over large datasets, DuckDB employs a columnar-vectorized query execution engine. This approach processes data in batches ("vectors"), significantly reducing overhead compared to traditional row-based systems like PostgreSQL, MySQL, or SQLite. DuckDB is also extensible, allowing users to define new data types, functions, file formats, and SQL syntax through an extension mechanism. Notably, support for popular formats like Parquet and JSON, as well as protocols like HTTP(S) and s3, are implemented as extensions.
DuckDB is extremely portable, capable of being compiled for all major operating systems (Linux, macOS, Windows) and CPU architectures (x86, ARM), and can even run in web browsers and on mobile phones via DuckDB-Wasm. It provides APIs for various programming languages, including Java, C, C++, Go, and Python. Despite its simplicity, DuckDB is feature-rich, offering extensive support for complex SQL queries, a large function library, window functions, transactional guarantees (ACID properties), persistent single-file databases, and secondary indexes. It is also deeply integrated into Python and R for efficient interactive data analysis.

For more technical details, visit the DuckDB Official Documentation.

DuckDB vs Apache Spark vs Trino: Quick Comparison

Feature	DuckDB	Apache Spark	Trino
Setup Costs	Free, lightweight	Cluster setup required	Requires distributed setup
Running Costs	Minimal	Ongoing cluster expenses	Cluster or cloud costs
Learning Curve	Basic SQL knowledge	Requires deeper learning	Moderate SQL knowledge
Performance	Fast for local queries	Scales well for big data	Optimized for distributed SQL
Use Case	Local analytics	Distributed data processing	Federated query engine

DuckDB is ideal for local analytical queries, Spark is better for large-scale distributed computing, and Trino excels at federated queries across multiple sources.

Hands-on: Unleashing DuckDB as Your Query Engine

In this tutorial, we’ll use DuckDB to query three different data sources:

PostgreSQL Database
MySQL Database
CSV Files
JSON from a Web Server

1. Setting Up the Environment

We’ll use Docker Compose to set up the infrastructure. First, clone the repository and start the services:

git clone https://github.com/mikekenneth/bp_duck_as_query_engine
cd bp_duck_as_query_engine
make up  # Start services

Check running containers with:

docker ps

Then, access DuckDB CLI:

make duckdb

To ease the setup, I created a duckdb_init.sql file that contains necessary SQL commands to connect to our external Data sources

2. Connecting to External Data Sources

a) Connecting to MinIO (s3-compatible Storage)

To interact with data stored in MinIO, we need to enable the HTTPS extension and configure the connection settings:

-- Enable the HTTPS Extension to connect to s3/Minio
INSTALL https;
LOAD https;

-- Set Connection settings
SET s3_region='us-east-1';
SET s3_url_style='path';
SET s3_endpoint='minio:9000';
SET s3_access_key_id='minio_root' ;
SET s3_secret_access_key='minio_toor';
SET s3_use_ssl=false; -- Needed when running without SSL

Let's verify if the settings are correctly applied.

b) Connecting to PostgreSQL

To query data within our PostgreSQL instance, we need to load the PostgreSQL extension and then attach to the database:

-- Load Extension
INSTALL postgres;
LOAD postgres;

-- Attach the PostgreSQL database
ATTACH 'dbname=postgres user=postgres password=postgres host=postgres port=5432' 
AS postgres_db (TYPE postgres, SCHEMA 'public');

Explanation: The INSTALL postgres; command adds support for interacting with PostgreSQL databases. The ATTACH command establishes a connection to the specified PostgreSQL database (dbname, user, password, host) and assigns it the alias postgres_db, making its tables accessible within DuckDB. We also specify the database type and schema.

We can now create a table in PostgreSQL by reading a CSV file from our MinIO instance:

-- Create the base table in Postgres from s3
create or replace table postgres_db.fct_trips as
(
    select *
    from read_csv("s3://duckdb-bucket/init_data/base_raw.csv")
);

c) Connecting to MySQL

-- Load Extension
INSTALL mysql;
LOAD mysql;

-- Attach the MYSQL database
ATTACH 'database=db user=user password=password host=mysql port=3306' 
AS mysql_db (TYPE mysql);

Explanation: The INSTALL mysql; command enables interaction with MySQL databases. The ATTACH command connects to the MySQL database using the provided credentials and assigns it the alias mysql_db. The database type is also specified.

Let's create a table in MySQL by reading a CSV file from MinIO:

-- Create the base table in MySQL from s3
create or replace table mysql_db.dim_credit_card as
(
    select *
    from read_csv("s3://duckdb-bucket/init_data/base_raw.csv")
);

Let's confirm that both databases are successfully attached:

d) Querying JSON from a Web Server

Finally, let's demonstrate fetching data directly from a web server serving a JSON file:

select distinct *
from read_json('http://nginx:80/companies_data.json')

This query directly reads and processes the JSON data available at the specified URL.

Here is the output of the query:

3. Querying Multiple Data Sources Simultaneously

With connectivity established to MinIO and our relational databases, we can now execute a single query that retrieves and joins data from all our configured sources:

---------------- Query multiple Data Sources Simultaneously ----------------
with 
  fct_trips as (
    select *
    from postgres_db.public.fct_trips
  ),
  dim_customer as (
    select distinct *
    from read_csv('s3://duckdb-bucket/init_data/customer.csv')
  ),
  dim_creditcard as (
    select distinct *
    from mysql_db.db.dim_credit_card
  ),
  dim_companies_data_web as (
    select distinct *
    from read_json('http://nginx:80/companies_data.json')
  )
select 
  trips."Trip ID",
  -- Customer info
  dcust.id as customer_id, dcust.name as customer_name,dcust.sex as customer_sex,
  dcust.address as customer_address, dcust.birth_date as customer_birth_date,
  -- Credit Card info
  dcard.credit_card_number as credit_card_number, dcard.expire_date as credit_card_expire_date,
  dcard.provider as credit_card_provider, dcard.owner_name as credit_card_owner_name,
  -- Company info
  dcompany.name as company_name, dcompany.address as company_address,
  dcompany.created_date as company_created_date, dcompany.num_of_employee as company_num_of_employee
from fct_trips trips
LEFT JOIN dim_customer dcust on trips.customer_id = dcust.id
LEFT JOIN dim_creditcard dcard on trips.credit_card_number = dcard.credit_card_number
LEFT JOIN dim_companies_data_web dcompany on trips.Company = dcompany.name;

Result:

4. Exporting Query Results

Furthermore, DuckDB allows direct export of query results to external storage, such as S3/MinIO, in various file formats. Here, we'll export the joined data as a Parquet file:

COPY (
with 
  fct_trips as (
    select *
    from postgres_db.public.fct_trips
...
...
...
from fct_trips trips
LEFT JOIN dim_customer dcust on trips.customer_id = dcust.id
LEFT JOIN dim_creditcard dcard on trips.credit_card_number = dcard.credit_card_number
LEFT JOIN dim_companies_data_web dcompany on trips.Company = dcompany.name

) TO 's3://duckdb-bucket/query_result.parquet' (FORMAT parquet);

We can then verify the exported Parquet file content directly with DuckDB

We can then validate the exported file:

SELECT * FROM read_parquet('s3://duckdb-bucket/query_result.parquet');

Conclusion

DuckDB is a powerful and lightweight query engine that allows querying multiple sources effortlessly. Whether working with CSV files, databases, or web-based JSON data, its simple SQL interface and high performance make it a fantastic choice for analytics. As the ecosystem grows, DuckDB continues to add more features, making it a strong alternative for local data processing.

References

Unlocking DuckDB from Anywhere - A Guide to Remote Access with Apache Arrow and Flight RPC (gRPC)

Mike Houngbadji — Thu, 12 Dec 2024 11:29:54 +0000

Be like a duck. Calm on the surface, but always paddling like the dickens underneath.
Michael Caine

Context

You’ve probably heard or used DuckDB in the Data Space. If not let’s introduce you to a game-changing tool that will dope your data stack as a Data Engineer / Analysts / Scientist / etc.

According to the official website DuckDB is an in-process SQL OLAP database management system. Simple, feature-rich, fast & open source., it’s simply said but forget to emphasize on it’s speed and flexibility. You read more here good_tutorial.

Here is my own definition of DuckDB: DuckDB is a SQL OLAP DB that brings together efficiency (V-fast vectorized query engine), performance and versatility/accessibility (easy install + can query files local or remotely + extensions that enable even more).

I know I sound like a DuckDB evangelist 😅, but please bear with me.

Now, let’s dive into what brings us together today!

Problem

DuckDB processes data using the local resources and the only (that i’m aware of at the moment) Cloud-like solution is MotherDuck (DuckDB’s own Cloud Service).

Hosting all tools and users on the same server in a data infrastructure has several drawbacks, which can significantly impact performance, reliability, security, and scalability such as:

Performance Bottlenecks: Tools and users competing for the same CPU, memory, and disk I/O can lead to resource contention and degraded performance during peak loads.
Single Point of Failure: If the server crashes or experiences an outage, all tools and users are affected simultaneously, leading to complete operational downtime.
Scalability Challenge: A single server has finite resources, making it difficult to scale as the number of users or tools increases.
Security Risks: A single compromised server can expose all tools and user data to potential breaches.
Operational Complexity: Diagnosing issues becomes harder when multiple tools and users are intertwined on a single server.
Compliance and Governance Risks: Hosting all users and tools together may violate regulatory requirements or customer expectations for data isolation.

Solution

What if you could let your DuckDB spread its wings ✈️ for broader access?
Remote connections open up opportunities to centralize data, enable multiple users, and achieve performance at scale.

In this guide, we’ll unlock the power of DuckDB’s remote access by pairing it with two key players: Apache Arrow (for efficient data transfer) and Arrow Flight RPC (gRPC under the hood) (for quick, reliable communication). By the end, you’ll have a fully functional API setup that will have your DuckDB connection flying in no time.

✈️ Let Your DuckDB Take Flight!

💻 Code

Here is the repository of the project:

https://github.com/mikekenneth/duckdb_streamlit_arrow_flight

Explanation

First, let’s discuss the High Level Design.
Below is Sequence Diagram of what we are trying to achieve:

The sequence diagram outlines the process of querying flight data via a web application. A User inputs an SQL query into the WebApp (Client), which sends a GetFlightInfo request to the FlightServer with the appropriate command descriptor. The server executes the SQL query using DuckDB, then returns a FlightInfo object containing a ticket to the client. The client subsequently uses the ticket to make a do_get request to the server, which streams the flight data back to the client. Finally, the client displays the results to the user.

To begin, we need the below components:

Python: We’ll use Python for it’s simplicity and accessibility.
DuckDB: Am sure I have yapped 😅enough about it.
Apache Arrow : It contains a set of technologies that enable big data systems to process and move data fast
Apache Arrow Flight RPC : Arrow Flight is an RPC framework for high-performance data services based on Arrow data, and is built on top of gRPC and the IPC format.
Streamlit: Streamlit turns data scripts into shareable web apps in minutes.

🏗️ Installation

To Install the project, we need follow the below steps:

Install the needed python modules

pip install -r requirements.txt

Create/Update the .env file with the below as needed:

# CAUTION: If using the source .env command,
## make sure there no space before & after the '=' signs in this file
# Flight Server
SERVER_FLIGHT_HOST='0.0.0.0'
SERVER_FLIGHT_PORT=8815
# If using Local Storage for the Flight datasets
SERVER_FLIGHT_DATA_DIR_TYPE='local'  # Options: ['local', 's3', 'minio']
SERVER_FLIGHT_DATA_DIR_BASE='data/datasets'
# DuckDB file
SERVER_DUCKDB_FILE='data/duck.db'

Server Code breakdown

For the Server, we used Apache Arrow Flight RPC to set up a gRPC Server that receives SQL queries as command.

`DuckDBClient`

This class exist to manage the connection and querying with DuckDB.

import duckdb
class DuckDBClient:
    def __init__(self, db_location: Path = None):
        if db_location is None:
            self.db_location = Path("data/duck.db")
        else:
            self.db_location = db_location
        self.db_location.parent.mkdir(exist_ok=True)
        self.duckdb_client = duckdb.connect(self.db_location)
    def query(self, sql_query) -> pa_table:
        result = self.duckdb_client.query(sql_query)
        return result.arrow()

`FlightServer`

Here we create the FlighServer class that holds all the methods/functionality of the Server.

import os
import pyarrow.parquet as pa_parquet
import pyarrow.flight as pa_flight
from pyarrow import Table as pa_table
from uuid import uuid4
from pathlib import Path
class FlightServer(pa_flight.FlightServerBase):
    def __init__(self, location="grpc://0.0.0.0:8815", repo: Path = Path("./datasets"), **kwargs):
        super(FlightServer, self).__init__(location, **kwargs)
        self._location = location
        self._repo = repo
    def _duckdb_query_handler(self, sql: str) -> str:
        duck_client = DuckDBClient()
        return duck_client.query(sql_query=sql)
    def _make_flight_info(self, dataset):
        dataset_path = self._repo / dataset
        schema = pa_parquet.read_schema(dataset_path)
        metadata = pa_parquet.read_metadata(dataset_path)
        descriptor = pa_flight.FlightDescriptor.for_path(dataset.encode("utf-8"))
        endpoints = [pa_flight.FlightEndpoint(dataset, [self._location])]
        return pa_flight.FlightInfo(schema, descriptor, endpoints, metadata.num_rows, metadata.serialized_size)
    def list_flights(self, context, criteria):
        for dataset in self._repo.iterdir():
            yield self._make_flight_info(dataset.name)
    def get_flight_info(self, context, descriptor: pa_flight):
        print(self._repo)
        if descriptor.descriptor_type == pa_flight.DescriptorType.CMD:
            command = descriptor.command.decode("utf-8")
            print(f"Executing the SQL command: {command}")
            query_result_arrow_table = self._duckdb_query_handler(sql=command)
            dataset_name = f"{str(uuid4())}.parquet" 
            pa_parquet.write_table(query_result_arrow_table, self._repo / dataset_name)
            print(f"Result Dataset: {dataset_name}")
            return self._make_flight_info(dataset=dataset_name)
        return self._make_flight_info(descriptor.path[0].decode("utf-8"))
    # TODO: You could use do_put to allow users to upload temporal tables      
    def do_put(self, context, descriptor, reader, writer):
        dataset = descriptor.path[0].decode("utf-8")
        dataset_path = self._repo / dataset
        data_table = reader.read_all()
        pa_parquet.write_table(data_table, dataset_path)
    def do_get(self, context, ticket):
        dataset = ticket.ticket.decode("utf-8")
        dataset_path = self._repo / dataset
        return pa_flight.RecordBatchStream(pa_parquet.read_table(dataset_path))
    def list_actions(self, context):
        return [("drop_dataset", "Delete a dataset.")]
    def do_drop_dataset(self, dataset):
        dataset_path = self._repo / dataset
        dataset_path.unlink()

`main()`

Below is the code the start the Flight Server.

if __name__ == "__main__":
    # Get Env Variables    FLIGHT_HOST = os.getenv("SERVER_FLIGHT_HOST", "0.0.0.0")
    FLIGHT_PORT = os.getenv("SERVER_FLIGHT_PORT", 8815)
    FLIGHT_DATA_DIR_TYPE = os.getenv("SERVER_FLIGHT_DATA_DIR_TYPE", "local")
    FLIGHT_DATA_DIR = os.getenv("SERVER_FLIGHT_DATA_DIR_BASE", "./datasets")
    print("Initiating Flight Server")
    server_location = f"grpc://{FLIGHT_HOST}:{FLIGHT_PORT}"
    server_data_dir = Path(FLIGHT_DATA_DIR)
    server_data_dir.mkdir(exist_ok=True)
    server = FlightServer(location=server_location, repo=server_data_dir)
    print("Starting Flight Server !")
    server.serve()

Client Code breakdown

On the client side, we use same Apache arrow library to connect to the Server, then use Streamlit to manage the web interface and display of results

Connect to Flight & Execute query

import os
import pyarrow.flight as pa_flight
from pyarrow import Table
# Get Env VariablesFLIGHT_HOST = os.getenv("SERVER_FLIGHT_HOST", "0.0.0.0")
FLIGHT_PORT = os.getenv("SERVER_FLIGHT_PORT", 8815)
def execute_query(sql_query: str) -> Table:
    # Connect to the Flight Server
    server_location = f"grpc://{FLIGHT_HOST}:{FLIGHT_PORT}"
    flight_client = pa_flight.connect(server_location)
    token_pair = flight_client.authenticate_basic_token(b'test', b'password')
    options = pa_flight.FlightCallOptions(headers=[token_pair])
    # Run query against the Flight Server
    flight_command_descriptor = pa_flight.FlightDescriptor.for_command(sql_query, options=options)
    # TODO: Add SQL inject parser    
    # Retrieve the Resulst Flight Descriptor
    response_flight = flight_client.get_flight_info(flight_command_descriptor)
    # Read content of the dataset
    reader = flight_client.do_get(response_flight.endpoints[0].ticket)
    result_table = reader.read_all()
    return response_flight, result_table

Streamlit Interface

Here we use Streamlit to build a simple interface that allow users to enter a SQL query (with streamlit-ace: a code block interface to ease the query written) and display results (schema, row count & data).

import streamlit as st
from streamlit_ace import st_ace
from xlib.flight import execute_query
st.set_page_config(page_icon=":duck:", page_title="Remote DuckDB ⛁  Client with Arrow Flight RPC")
st.title("Remote DuckDB ⛁  Client with Arrow Flight")
with st.form(key="query_form"):
    st.markdown("## 🇸 🇶 🇱 Query Space")
    sql_query = st_ace(
        # value="select '1' as testing;"        placeholder="Write your Query here...",
        language="sql",
        theme="dracula",
        keybinding="vscode",
        font_size=14,
        tab_size=4,
        show_gutter=True,
        show_print_margin=False,
        wrap=False,
        auto_update=True,
        readonly=False,
        min_lines=20,
        key="code_content",
    )
    submit = st.form_submit_button(label="Run Query")
if submit:
    # Run the Query against Flight Server
    result_flight_info, result_table = execute_query(sql_query)
    st.subheader("Results", divider="rainbow", anchor="results")
    tab1, tab2, tab3, tab4 = st.tabs(["Result", "Query", "Schema", "Info"])
    with tab1:
        st.dataframe(result_table.to_pandas(), use_container_width=True)
    with tab2:
        st.markdown(f"```
{% endraw %}
sql\n {sql_query} \n
{% raw %}
```")
    with tab3:
        st.markdown(f"```
{% endraw %}
csharp\n=== Result Schema ===\n{result_flight_info.schema}\n=====================\n
{% raw %}
```")
    with tab4:
        st.dataframe(
            {
                "Total Rows:": result_flight_info.total_records,
                "Total Size (Bytes):": result_flight_info.total_bytes,
                "Dataset Paths:": [i.decode("utf-8") for i in result_flight_info.descriptor.path],
            },
            hide_index=True,
        )

📹 Demo

Here is a short demo of the solution.

https://www.youtube.com/watch?v=jS5p8pwnlOo

➡️ Next Steps

As the quote says Nothing is perfect.
Hence, I will adding more features to the project as time goes by and am very open to contributions, comments, etc.

🙏Thanks

Thanks for reading.

Apache Spark SQL / Hive: Create External Table based on File in HDFS

Mike Houngbadji — Thu, 13 Jul 2023 10:30:48 +0000

To create a Table based on a file located in HDFS, we'll proceed as follow:

Update the file/folder to HDFS:

hadoop fs -put /local/source/location /hdfs/destination/location

Create the table using the below SQL:

CREATE TABLE sample_table(
        key STRING,
        data STRING)
USING CSV  -- This is based on the format of your source files
OPTIONS ('delimiter'=',',  -- This only needed for delimited file.
        'path'='hdfs:///hdfs/destination/location')

We can the now query our table:

SELECT *
FROM sample_table

References:
SparkSQL Documentation - Create Table

PS:
I wrote this to also help myself retrieve the solution faster.

Building a Data Lakehouse for Analyzing Elon Musk Tweets using MinIO, Apache Airflow, Apache Drill and Apache Superset

Mike Houngbadji — Fri, 20 Jan 2023 23:50:10 +0000

Every act of conscious learning requires the willingness to suffer an injury to one's self-esteem. That is why young children, before they are aware of their own self-importance, learn so easily.
Thomas Szasz

Motivation

A Data Lakehouse is a modern data architecture that combines the scalability and flexibility of a data lake with the governance and performance of a data warehouse. This approach allows organizations to store and analyze large amounts of structured and unstructured data in a single platform, enabling more efficient and effective data-driven decision making.

To deep dive in this, we will build a Data lakehouse solution for analyzing tweets from Elon Musk using MinIO as storage, Apache Drill as query engine, Apache Superset for visualization and Apache Airflow for orchestration. This article will take you through the process of building and utilizing this solution to gain insights and make data-driven decisions.

Then, we’ll use Docker-Compose to easily deploy our solution.

At the end of this project, we'll be able to view the below dashboard of Elon's Tweets with some useful metrics:

Architecture

Table of Content

What is Apache Airflow?
What is MinIO ?
What is Apache Drill ?
What is Apache Superset ?
Code
- get_twitter_data()
- clean_twitter_data()
- write_to_bucket()
- DAG (Direct Acyclic Graph)
- Apache Drill Configuration
- Apache Superset Configuration
- docker-compose & .env files
Results

What is Apache Airflow

Airflow is a platform created by the community to programmatically author, schedule and monitor workflows.

Apache Airflow is an open-source workflow orchestration written in Python. It uses DAG (Direct Acyclic Graphs) to represent workflows. It is highly customizable/flexible and have a quite active community.

In our project, we'll use Apache-Airflow for orchestrating the job that pull data from Twitter API and write to MinIO (s3 compatible object storage).

💡 You can read more [here](https://airflow.apache.org/).

What is MinIO

MinIO offers high-performance, S3 compatible object storage.

MinIO is an open-source Multi-cloud Object Storage and fully compatible with AWS s3. With MinIO you can host your on-premises or cloud Object storage.

Here, we are using MinIO as an alternative to AWS s3 to store the tweets data as objects in buckets.

💡 You can read more [here](https://min.io/).

What is Apache Drill

Schema-free SQL Query Engine for Hadoop, NoSQL and Cloud Storage.

Apache-Drill is a Query engine that enables querying data directly from multiple sources in a single query. It allows to query data where it's (the data) sits without having to load into a warehouse. It's supports many sources like AWS s3, RDBMS, MongoDB, Druid, etc.

Hence, with Drill's capability of querying data directly from and s3 buckets, we'll use it to query the tweets data directly from MinIO without having to load into a warehouse.

💡 You can read more [here](https://drill.apache.org/).

What is Apache Superset

Apache Superset is a modern data exploration and visualization platform.

Superset is fast, lightweight, intuitive, and loaded with options that make it easy for users of all skill sets to explore and visualize their data, from simple line charts to highly detailed geospatial charts.

We will use Superset for visualizing charts and get insights on the tweets data stored in MinIO Buckets directly using Apache-Drill Query Engine.

💡 You can read more [here](https://superset.apache.org/).

Code

The full code can be accessed here:

Source code:

https://github.com/mikekenneth/twitter_data-lakehouse_minio_drill_superset

get_twitter_data()

Below is the python Task that pulls Elon’s tweets from Twitter API into a Python list:

import os
import requests
from airflow.decorators import dag, task

@task
def get_twitter_data():
    TWITTER_BEARER_TOKEN = os.getenv("TWITTER_BEARER_TOKEN")

    # Get tweets using Twitter API v2 & Bearer Token
    BASE_URL = "https://api.twitter.com/2/tweets/search/recent"
    USERNAME = "elonmusk"
    FIELDS = {"created_at", "lang", "attachments", "public_metrics", "text", "author_id"}

    url = f"{BASE_URL}?query=from:{USERNAME}&tweet.fields={','.join(FIELDS)}&expansions=author_id&max_results=50"
    response = requests.get(url=url, headers={"Authorization": f"Bearer {TWITTER_BEARER_TOKEN}"})
    response_content = response.json()

    data = response_content["data"]
    includes = response_content["includes"]
    return data, includes

clean_twitter_data()

Below is the python Task that cleans & transforms the tweets in the adequate format:

from uuid import uuid4
from datetime import datetime
from airflow.decorators import dag, task

@task
def clean_twitter_data(tweets_data):
    data, includes = tweets_data

    batchId = uuid4().hex
    batchDatetime = datetime.now().strftime("%Y-%m-%d %H:%M:%S")

    # Refine tweets data
    tweet_list = []
    for tweet in data:
        refined_tweet = {
            "tweet_id": tweet["id"],
            "username": includes["users"][0]["username"],  # Get username from the included data
            "user_id": tweet["author_id"],
            "text": tweet["text"],
            "like_count": tweet["public_metrics"]["like_count"],
            "retweet_count": tweet["public_metrics"]["retweet_count"],
            "created_at": tweet["created_at"],
            "batchID": batchId,
            "batchDatetime": batchDatetime,
        }
        tweet_list.append(refined_tweet)
    return tweet_list, batchDatetime, batchId

write_to_bucket()

Below is the python task transforms the tweets list into a Pandas Dataframe, then writes it in our MinIO Object Storage as a Parquet file.

import os
from airflow.decorators import dag, task

@task
def write_to_bucket(data):
    tweet_list, batchDatetime_str, batchId = data
    batchDatetime = datetime.strptime(batchDatetime_str, "%Y-%m-%d %H:%M:%S")

    import pandas as pd
    from minio import Minio
    from io import BytesIO

    MINIO_BUCKET_NAME = os.getenv("MINIO_BUCKET_NAME")
    MINIO_ROOT_USER = os.getenv("MINIO_ROOT_USER")
    MINIO_ROOT_PASSWORD = os.getenv("MINIO_ROOT_PASSWORD")

    df = pd.DataFrame(tweet_list)
    file_data = df.to_parquet(index=False)

    client = Minio("minio:9000", access_key=MINIO_ROOT_USER, secret_key=MINIO_ROOT_PASSWORD, secure=False)

    # Make MINIO_BUCKET_NAME if not exist.
    found = client.bucket_exists(MINIO_BUCKET_NAME)
    if not found:
        client.make_bucket(MINIO_BUCKET_NAME)
    else:
        print(f"Bucket '{MINIO_BUCKET_NAME}' already exists!")

    # Put parquet data in the bucket
    filename = (
        f"tweets/{batchDatetime.strftime('%Y/%m/%d')}/elon_tweets_{batchDatetime.strftime('%H%M%S')}_{batchId}.parquet"
    )
    client.put_object(
        MINIO_BUCKET_NAME, filename, data=BytesIO(file_data), length=len(file_data), content_type="application/csv"
    )

We used the below object nomenclature to create sub-folders to partition the data by year, month & day:

# Result: filename = "tweets/2023/01/10/elon_tweets_112512_6c0d2558ff464091b9daf75c74f28a51.parquet"
filename = f"tweets/{batchDatetime.strftime('%Y/%m/%d')}/elon_tweets_{batchDatetime.strftime('%H%M%S')}_{batchId}.parquet"

DAG (Direct Acyclic Graph)

Below is the DAG itself that allows specifying the dependencies between tasks and define the scheduling interval (in our case, every hour) of the DAG:

from datetime import datetime
from airflow.decorators import dag, task

@dag(
    schedule="0 * * * *",
    start_date=datetime(2023, 1, 10),
    catchup=False,
    tags=["twitter", "etl"],
)
def twitter_etl():
    raw_data = get_twitter_data()
    cleaned_data = clean_twitter_data(raw_data)
    write_to_bucket(cleaned_data)

twitter_etl()

Apache Drill Configuration

To grant Apache Drill access to our MinIO (s3) bucket, we need to defining Access Keys in the Drill core-site.xml file:

<?xml version="1.0" encoding="UTF-8" ?>
<configuration>

    <property>
        <name>fs.s3a.access.key</name>
        <value>minioadmin</value>
    </property>

    <property>
        <name>fs.s3a.secret.key</name>
        <value>minioadmin</value>
    </property>

    <property>
      <name>fs.s3a.endpoint</name>
      <value>http://minio:9000</value>
    </property>

    <property>
      <name>fs.s3a.connection.ssl.enabled</name>
      <value>false</value>
    </property>

    <property>
      <name>fs.s3a.path.style.access</name>
      <value>true</value>
    </property>

</configuration>

Next, we’ll configure the S3 Storage Plugin to specify the bucket (twitter-data) to be access by Apache Drill in the storage-plugins-override.conf file:

"storage": {
  s3: {
    type: "file",
    connection: "s3a://twitter-data",
    workspaces: {
      "root": {
        "location": "/",
        "writable": false,
        "defaultInputFormat": null,
        "allowAccessOutsideWorkspace": false
      }
    },
    formats: {
      "parquet": {
        "type": "parquet"
      },
      "csv" : {
        "type" : "text",
        "extensions" : [ "csv" ]
      }
    },
    enabled: true
  }
}

💡 You can read more here.

Apache Superset Configuration

To be able to query Apache-Drill, we need to build a custom image of apache/superset using superset_drill.Dockerfile in which we'll install the sqlalchemy-drill module the enables the interactions with Apache-Drill:

FROM apache/superset
# Switching to root to install the required packages
USER root
# install requirements for Apache Drill
RUN pip install sqlalchemy-drill
# Switching back to using the `superset` user
USER superset

💡 You can read more here.

docker-compose & .env files

Below is the .env file that we need to create that hold the environmental variables needed to run our pipeline:

💡 You can read this to learn our to generate the TWITTER_BEARER_TOKEN.

# Twitter
TWITTER_BEARER_TOKEN="TOKEN"

# Minio
MINIO_ROOT_USER=minioadmin
MINIO_ROOT_PASSWORD=minioadmin
MINIO_BUCKET_NAME='twitter-data'

# Superset
SUPERSET_USERNAME=admin
SUPERSET_PASSWORD=admin

# Apache Airflow
## Meta-Database
POSTGRES_USER=airflow
POSTGRES_PASSWORD=airflow
POSTGRES_DB=airflow

## Airflow Core
AIRFLOW__CORE__FERNET_KEY=''
AIRFLOW__CORE__EXECUTOR=LocalExecutor
AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION=True
AIRFLOW__CORE__LOAD_EXAMPLES=False
AIRFLOW_UID=50000
AIRFLOW_GID=0

## Backend DB
AIRFLOW__DATABASE__SQL_ALCHEMY_CONN=postgresql+psycopg2://airflow:airflow@postgres/airflow
AIRFLOW__DATABASE__LOAD_DEFAULT_CONNECTIONS=False

## Airflow Init
_AIRFLOW_DB_UPGRADE=True
_AIRFLOW_WWW_USER_CREATE=True
_AIRFLOW_WWW_USER_USERNAME=airflow
_AIRFLOW_WWW_USER_PASSWORD=airflow
_PIP_ADDITIONAL_REQUIREMENTS= "minio pandas requests"

And below is the docker-compose.yaml file that allow to spin up the needed infrastructure for our pipeline:

version: '3.4'

x-common:
  &common
  image: apache/airflow:2.5.0
  user: "${AIRFLOW_UID}:0"
  env_file:
    - .env
  volumes:
    - ./app/dags:/opt/airflow/dags
    - ./app/logs:/opt/airflow/logs

x-depends-on:
  &depends-on
  depends_on:
    postgres:
      condition: service_healthy
    airflow-init:
      condition: service_completed_successfully

services:
  minio:
    image: minio/minio:latest
    ports:
      - '9000:9000'
      - '9090:9090'
    volumes:
      - './data:/data'
    env_file:
      - .env
    command: server --console-address ":9090" /data
    healthcheck:
      test:
        [
          "CMD",
          "curl",
          "-f",
          "http://localhost:9000/minio/health/live"
        ]
      interval: 30s
      timeout: 20s
      retries: 3

  drill:
    env_file:
      - .env
    image: apache/drill:latest
    ports:
      - '8047:8047'
      - '31010:31010'
    volumes:
      # If needed, override default settings
      - ./conf/drill/core-site.xml:/opt/drill/conf/core-site.xml
      # Register default storage plugins
      - ./conf/drill/storage-plugins-override.conf:/opt/drill/conf/storage-plugins-override.conf
    stdin_open: true
    tty: true

  superset_drill:
    env_file:
      - .env
    ports:
      - '8088:8088'
    build:
      context: .
      dockerfile: superset_drill.Dockerfile
    volumes:
      - ./superset.db:/app/superset_home/superset.db

  postgres:
    image: postgres:13
    container_name: postgres
    ports:
      - "5433:5432"
    healthcheck:
      test: [ "CMD", "pg_isready", "-U", "airflow" ]
      interval: 5s
      retries: 5
    env_file:
      - .env

  scheduler:
    <<: *common
    <<: *depends-on
    container_name: airflow-scheduler
    command: scheduler
    restart: on-failure
    ports:
      - "8793:8793"

  webserver:
    <<: *common
    <<: *depends-on
    container_name: airflow-webserver
    restart: always
    command: webserver
    ports:
      - "8080:8080"
    healthcheck:
      test:
        [
          "CMD",
          "curl",
          "--fail",
          "http://localhost:8080/health"
        ]
      interval: 30s
      timeout: 30s
      retries: 5

  airflow-init:
    <<: *common
    container_name: airflow-init
    entrypoint: /bin/bash
    command:
      - -c
      - |
        mkdir -p /sources/logs /sources/dags
        chown -R "${AIRFLOW_UID}:0" /sources/{logs,dags}
        exec /entrypoint airflow version

Results

When we access the Apache-Airflow Web UI, we can see the DAG and we can run it directly to see the results.
DAG (Apache-Airflow web UI):
Tweets Parquet file generated in our bucket (MinIO Console):
If needed, we can query our Data-Lakehouse directly using Apache-Drill Web Interface:
Finally, we can visualize our Dashboard (I built this already, but it can be easily modified and the data is stored in the superset.db file):

This is a wrap. I hope this helps you.

About Me

I am a Data Engineer with 3+ years of experience and more years as a Software Engineer (5+ years). I enjoy learning and teaching (mostly learning 😎).

You can get in touch with me through Twitter & LinkedIn or mike.kenneth47@gmail.com.

Twitter Data Pipeline with Apache Airflow + MinIO (S3 compatible Object Storage)

Mike Houngbadji — Fri, 06 Jan 2023 23:20:43 +0000

Twitter Data Pipeline with Apache Airflow + MinIO (S3 compatible Object Storage)

The more that you read, the more things you will know. The more that you learn, the more places you’ll go.
Dr. Seuss

Motivation

During my journey as a Data Engineer, I stumbled upon many tools.
One that caught my attention is MinIO, a Multi-cloud Object Storage that is AWS s3 Compatible.

To learn more about it, I built a Data Pipeline that uses Apache Airflow to pull Elon Musk tweets using the Twitter API and store the result in a CSV stored in a MinIO (OSS alternative to AWS s3) Object Storage bucket.

Then, we’ll use Docker-Compose to easily deploy our code.

Table of Content

What is Apache Airflow?
What is MinIO ?
Code
- get_twitter_data()
- dump_data_to_bucket()
- DAG (Direct Acyclic Graph)
- docker-compose & .env files

What is Apache Airflow

Airflow is a platform created by the community to programmatically author, schedule and monitor workflows.

Apache Airflow is an opensource workflow orchestration written in Python. It uses DAG (Direct Acyclic Graphs) to represent workflows. It is highly customizable/flexible and have a quite active community.

You can read more here.

What is MinIO

MinIO offers high-performance, S3 compatible object storage.

MinIO is an opensource Multi-cloud Object Storage and fully compatible with AWS s3. With MinIO you can host your own on-premises or cloud Object Storage.

You can read more here.

Code

The full code can be accessed.

Source code:

https://github.com/mikekenneth/airflow_minio_twitter_data_pipeline

get_twitter_data()

Below is the python Task that pulls Elon’s tweets from Twitter API into a python list:

import os
import json
import requests
from airflow.decorators import dag, task

@task
def get_twitter_data():
    TWITTER_BEARER_TOKEN = os.getenv("TWITTER_BEARER_TOKEN")

    # Get tweets using Twitter API v2 & Bearer Token
    BASE_URL = "https://api.twitter.com/2/tweets/search/recent"
    USERNAME = "elonmusk"
    FIELDS = {"created_at", "lang", "attachments", "public_metrics", "text", "author_id"}

    url = f"{BASE_URL}?query=from:{USERNAME}&tweet.fields={','.join(FIELDS)}&expansions=author_id&max_results=50"
    response = requests.get(url=url, headers={"Authorization": f"Bearer {TWITTER_BEARER_TOKEN}"})
    response = json.loads(response.content)

    data = response["data"]
    includes = response["includes"]

    # Refine tweets data
    tweet_list = []
    for tweet in data:
        refined_tweet = {
            "tweet_id": tweet["id"],
            "username": includes["users"][0]["username"],  # Get username from the included data
            "user_id": tweet["author_id"],
            "text": tweet["text"],
            "like_count": tweet["public_metrics"]["like_count"],
            "retweet_count": tweet["public_metrics"]["retweet_count"],
            "created_at": tweet["created_at"],
        }
        tweet_list.append(refined_tweet)
    return tweet_list

dump_data_to_bucket()

Below is the python Task that transforms the tweets list into a Pandas dataframe, then dumps it in our MinIO Object Storage as a CSV file:

import os
from airflow.decorators import dag, task

@task
def dump_data_to_bucket(tweet_list: list):
    import pandas as pd
    from minio import Minio
    from io import BytesIO
    MINIO_BUCKET_NAME = os.getenv("MINIO_BUCKET_NAME")
    MINIO_ROOT_USER = os.getenv("MINIO_ROOT_USER")
    MINIO_ROOT_PASSWORD = os.getenv("MINIO_ROOT_PASSWORD")

    df = pd.DataFrame(tweet_list)
    csv = df.to_csv(index=False).encode("utf-8")

    client = Minio("minio:9000", access_key=MINIO_ROOT_USER, secret_key=MINIO_ROOT_PASSWORD, secure=False)

    # Make MINIO_BUCKET_NAME if not exist.
    found = client.bucket_exists(MINIO_BUCKET_NAME)
    if not found:
        client.make_bucket(MINIO_BUCKET_NAME)
    else:
        print(f"Bucket '{MINIO_BUCKET_NAME}' already exists!")

    # Put csv data in the bucket
    client.put_object(
        "airflow-bucket", "twitter_elon_musk.csv", data=BytesIO(csv), length=len(csv), content_type="application/csv"
    )

DAG (Direct Acyclic Graph)

Below is the DAG itself that allows specifying the dependencies between tasks:

from datetime import datetime
from airflow.decorators import dag, task

@dag(
    schedule="0 */2 * * *",
    start_date=datetime(2022, 12, 26),
    catchup=False,
    tags=["twitter", "etl"],
)
def twitter_etl():
    dump_data_to_bucket(get_twitter_data())

twitter_etl()

docker-compose & .env files

Below is the .env file that we need to create that hold the environmental variables needed to run our pipeline:

You can read this to learn our to generate the TWITTER_BEARER_TOKEN.

# Twitter (Must not be empty)
TWITTER_BEARER_TOKEN=""

# Meta-Database
POSTGRES_USER=airflow
POSTGRES_PASSWORD=airflow
POSTGRES_DB=airflow

# Airflow Core
AIRFLOW__CORE__FERNET_KEY=''
AIRFLOW__CORE__EXECUTOR=LocalExecutor
AIRFLOW__CORE__DAGS_ARE_PAUSED_AT_CREATION=True
AIRFLOW__CORE__LOAD_EXAMPLES=False
AIRFLOW_UID=50000
AIRFLOW_GID=0

# Backend DB
AIRFLOW__DATABASE__SQL_ALCHEMY_CONN=postgresql+psycopg2://airflow:airflow@postgres/airflow
AIRFLOW__DATABASE__LOAD_DEFAULT_CONNECTIONS=False

# Airflow Init
_AIRFLOW_DB_UPGRADE=True
_AIRFLOW_WWW_USER_CREATE=True
_AIRFLOW_WWW_USER_USERNAME=airflow
_AIRFLOW_WWW_USER_PASSWORD=airflow
_PIP_ADDITIONAL_REQUIREMENTS= "minio pandas requests"

# Minio
MINIO_ROOT_USER=minio_user
MINIO_ROOT_PASSWORD=minio_password123
MINIO_BUCKET_NAME='airflow-bucket'

And below is the docker-compose.yaml file that allow to spin up the needed infrastructure for our pipeline:

version: '3.4'

x-common:
  &common
  image: apache/airflow:2.5.0
  user: "${AIRFLOW_UID}:0"
  env_file:
    - .env
  volumes:
    - ./app/dags:/opt/airflow/dags
    - ./app/logs:/opt/airflow/logs

x-depends-on:
  &depends-on
  depends_on:
    postgres:
      condition: service_healthy
    airflow-init:
      condition: service_completed_successfully

services:
  minio:
    image: minio/minio:latest
    ports:
      - '9000:9000'
      - '9090:9090'
    volumes:
      - './minio_data:/data'
    env_file:
      - .env
    command: server --console-address ":9090" /data

  postgres:
    image: postgres:13
    container_name: postgres
    ports:
      - "5433:5432"
    healthcheck:
      test: [ "CMD", "pg_isready", "-U", "airflow" ]
      interval: 5s
      retries: 5
    env_file:
      - .env

  scheduler:
    <<: *common
    <<: *depends-on
    container_name: airflow-scheduler
    command: scheduler
    restart: on-failure
    ports:
      - "8793:8793"

  webserver:
    <<: *common
    <<: *depends-on
    container_name: airflow-webserver
    restart: always
    command: webserver
    ports:
      - "8080:8080"
    healthcheck:
      test:
        [
          "CMD",
          "curl",
          "--fail",
          "http://localhost:8080/health"
        ]
      interval: 30s
      timeout: 30s
      retries: 5

  airflow-init:
    <<: *common
    container_name: airflow-init
    entrypoint: /bin/bash
    command:
      - -c
      - |
        mkdir -p /sources/logs /sources/dags
        chown -R "${AIRFLOW_UID}:0" /sources/{logs,dags}
        exec /entrypoint airflow version

When we access the Apache-Airflow Web UI, we can see the DAG and we can run it directly to see the results.

DAG (Apache-Airflow web UI):

Tweets file generated in our bucket (MinIO Console):

This is a wrap. I hope this helps you.

About Me

I am a Data Engineer with 3+ years of experience and more years as a Software Engineer (5+ years). I enjoy learning and teaching (mostly learning 😎).

You can get in touch with me by mike.kenneth47@gmail.com, Twitter & LinkedIn.

Article posted using bloggu.io. Try it for free.

Streamlit + DuckDB Tutorial

Mike Houngbadji — Fri, 25 Nov 2022 22:00:44 +0000

Streamlit + DuckDB Tutorial

The beautiful thing about learning is that nobody can take it away from you.
B.B King

Motivation

In this era of Everything Web, we are basically compelled to build web interfaces for our various application to ease the access, hence, bring value.

As a Python developer , we have a lot modules & framework that helps us build web interfaces.

This is where Streamlit comes in, with it’s pythonic way of creating and web applications, it's the go to solution to build fast and with great ease web application.

We’ll use duckDB for it’s built-in OLAP capabilities and easy usage (the databases is a file, no complex infrastructure needed).

In this tutorial, you’ll learn how to build web apps using Streamlit and using duckDB as your database.

Table of content

What is Streamlit ?
What is duckDB ?
Code
- Data Generator
- Extract data from DuckDB
- Visualization
- Filtering Data
- Download / Exporting Data

What is Streamlit ?

Streamlit turns data scripts into shareable web apps in minutes.

All in pure Python. No front‑end experience required.

With Streamlit, we can easily create using python ONLY dynamic web application. It’s ideal for data apps as it natively has built-in function to present data in different ways (tables, charts, etc.).

You can read more here

What is duckDB ?

DuckDB is an in-process SQL OLAP database management system.

It is serverless comes with great features like (Direct Parquet & CSV querying, Vectorized Engine, Parallel query Processing). It is an alternative to SQLite purposely built for Analytics workload.

You can read more here.

Code

The full code can be access below.

https://github.com/mikekenneth/streamlit_duckdb

Data Generator

The below generates a 1000 rows data and loads to orders table in duck.db when the button is clicked:

from generate_data import generate_dataset_orders, load_file

button = st.button(label="Generate or Refresh Data")
if button:
    db = "duck.db"
    destination_table_name = "orders"
    filename = "orders.csv"
    generate_dataset_orders(filename=filename, num_rows=1000)
    load_file(db=db, infile_path=filename, table_name=destination_table_name)

Screenshot

I wrote a little helper to easily generate the orders data and load into duckDB. The code can be accessed here.

Extract data from DuckDB

I created a function execute_query to easily get the data using only the desired SQL query. It also allows specifying the type of output needed. You can read more in the official documentation.

def execute_query(query: str, return_type: str = "df"):
    with duckdb.connect("duck.db", read_only=True) as con:
        if return_type == "df":
            return con.execute(query).df()
        elif return_type == "arrow":
            return con.execute(query).arrow()
        elif return_type == "list":
            return con.execute(query).fetchall()

destination_table_name = "orders"
data = execute_query(f"select * from {destination_table_name}", return_type="df")

Display a sample of the data

With Streamlit we can easily display a Dataframe like a table. You can read more here.

st.write("## Sample")
st.dataframe(data.head(10), height=300)

Screenshot

Visualization

Streamlit allow us to easily display charts. We will use it to display some bar charts. You can read more here.

st.write("## Visualization")
option = st.selectbox("Select a dimension", ["product_name", "customer_name", "status"], key="option")
if option:
    st.write(f"### Bar Chart: {option} x Quantity")
    st.bar_chart(data, x=option, y="quantity")

    st.write(f"### Bar Chart: {option} x Amount")
    st.bar_chart(data, x=option, y="amount")

    st.write(f"### Bar Chart: {option} x Count")
    st.bar_chart(data[option].value_counts())

Screenshot

Filtering Data

We can also filter out data directly from the web using the below. You can read more here.

st.write("## Filters (by Products Name)")
products_list = [
    row[0]
    for row in execute_query(
        f"select distinct(product_name) from {destination_table_name}", db=db, return_type="list"
    )
]
product_filter = st.selectbox(label="Select a Product", options=products_list, key="product_filter")
if product_filter != "--":
    result = execute_query(
        f"select * from {destination_table_name} where product_name = '{product_filter}'",
        db=db,
        return_type="df",
    )
    st.dataframe(result, height=400)

Screenshot

Download / Exporting Data

We can export data to csv using the Streamlit st.download_button function.

@st.experimental_memo  # An optimization wrapper to memoize the result of the function
def export_df(df):
    return df.to_csv(index=False).encode("utf-8")

st.write("### Download Data")
st.download_button(
    label="Press to Download",
    data=export_df(result),
    file_name=f"orders - product='{product_filter}'.csv",
    mime="text/csv",
    key="download-csv",
)

Screenshot

That is a wrap. I hope this helps you. 🙂

About Me

I am a Data Engineer with 2+ years of experience and more years as a Software Engineer (5+ years). I enjoy learning and teaching (mostly learning 😎).

You can get in touch with me by email, Twitter & LinkedIn.

Apache Spark SQL: CTAS USING CSV with specific delimiter

Mike Houngbadji — Wed, 16 Nov 2022 17:53:15 +0000

To use CTAS (Create table as Select) in Apache Spark with a specific delimiter using the OPTION syntax:

CREATE TABLE sample_table
USING CSV
OPTIONS ('delimiter'='|')
AS
SELECT ....

SparkSQL Documentation

PS:
I wrote this to also help myself retrieve the solution faster.