Goodreads ELT pipelines

git clone https://github.com/lelouvincx/goodreads-elt-pipeline.git project
cd project

touch .env
cp env.template .env
touch .spark_master.env
cp spark_master.env.template .spark_master.env
touch .spark_worker.env
cp spark_worker.env.template .spark_worker.env

# MySQL
MYSQL_HOST=de_mysql
MYSQL_PORT=3306
MYSQL_DATABASE=goodreads
MYSQL_USER=admin
MYSQL_PASSWORD=admin123
MYSQL_ROOT_PASSWORD=root123

# PostgreSQL
POSTGRES_HOST=de_psql
POSTGRES_PORT=5432
POSTGRES_USER=admin
POSTGRES_PASSWORD=admin123
POSTGRES_DB=goodreads
POSTGRES_HOST_AUTH_METHOD=trust

# Google Drive
GDRIVE_CLIENT_SECRET_FILE=client_secret.json
GDRIVE_PICKLE_FILE=token_drive_v3.pickle
GDRIVE_API_NAME=drive
GDRIVE_API_VERSION=v3
GDRIVE_SCOPES=https://www.googleapis.com/auth/drive.readonly

# Dagster
DAGSTER_PG_HOSTNAME=de_psql
DAGSTER_PG_USERNAME=admin
DAGSTER_PG_PASSWORD=admin123
DAGSTER_PG_DB=postgres
DAGSTER_OVERALL_CONCURRENCY_LIMIT=1
DAGSTER_HOME=/opt/dagster/dagster_home

# dbt
DBT_HOST=de_psql
DBT_USER=admin
DBT_PASSWORD=admin123
DBT_DATABASE=goodreads
DBT_SCHEMA=recommendations
# MinIO
MINIO_ENDPOINT=minio:9000
MINIO_ROOT_USER=minio
MINIO_ROOT_PASSWORD=minio123
MINIO_ACCESS_KEY=minio
MINIO_SECRET_KEY=minio123
DATALAKE_BUCKET=lakehouse
AWS_ACCESS_KEY_ID=minio
AWS_SECRET_ACCESS_KEY=minio123
AWS_REGION=us-east-1

# MinIO client (mc)
AWS_ACCESS_KEY_ID=minio
AWS_SECRET_ACCESS_KEY=minio123
AWS_REGION=us-east-1

# Spark
SPARK_MASTER_URL=spark://spark-master:7077
SPARK_VERSION=3.3.2
HADOOP_VERSION=3

# Metabase
MB_DB_TYPE=postgres
MB_DB_DBNAME=goodreads
MB_DB_PORT=5432
MB_DB_USER=admin
MB_DB_PASS=admin123
MB_DB_HOST=de_psql
MB_DB_FILE=/metabase_data/metabase.db

# DO NOT RUN BOTH BELOW COMMANDS, ONLY CHOOSE ONE
# Setup python environment
pipenv install
# Or create virtualenv and install manually by requirements.txt
make install

# Build docker images
make build-dagster
make build-spark
make build-pipeline
make build-streamlit

# Run containers dettached
make up-bg

# Check running containers
docker compose ps -a

# Check code quality
make check
make lint

# Format pipelines
black ./elt_pipeline

# Test coverage
make test

make to_mysql_root

SET GLOBAL local_infile=TRUE;
-- Check if local_infile was turned on
SHOW VARIABLES LIKE "local_infile";
exit

# Create tables with schema
make mysql_create

# Load csv into created tables
make mysql_load

make psql_create

def connect_mysql(config) -> str:
    conn_info = (
        f"mysql://{config['user']}:{config['password']}"
        + f"@{config['host']}:{config['port']}"
        + f"/{config['database']}"
    )
    return conn_info

def extract_data(self, sql: str) -> pl.DataFrame:
    """
    Extract data from MySQL database as polars DataFrame
    """
    conn_info = connect_mysql(self._config)
    df_data = pl.read_database(query=sql, connection_uri=conn_info)
    return df_data

COMPUTE_KIND = "SQL"
LAYER = "bronze"
YEARLY = StaticPartitionsDefinition(
    [str(year) for year in range(1975, datetime.today().year)]
)

# book from my_sql
@asset(
    description="Load table 'book' from MySQL database as polars DataFrame, and save to minIO",
    partitions_def=YEARLY,
    io_manager_key="minio_io_manager",
    required_resource_keys={"mysql_io_manager"},
    key_prefix=["bronze", "goodreads"],
    compute_kind=COMPUTE_KIND,
    group_name=LAYER,
)
def bronze_book(context) -> Output[pl.DataFrame]:
    query = "SELECT * FROM book"
    try:
        partion_year_str = context.asset_partition_key_for_output()
        partition_by = "PublishYear"
        query += f" WHERE {partition_by} = {partion_year_str}"
        context.log.info(f"Partition by {partition_by} = {partion_year_str}")
    except Exception:
        context.log.info("No partition key found, full load data")

    df_data = context.resources.mysql_io_manager.extract_data(query)
    context.log.info(f"Table extracted with shape: {df_data.shape}")

    return Output(
        value=df_data,
        metadata={
            "table": "book",
            "row_count": df_data.shape[0],
            "column_count": df_data.shape[1],
            "columns": df_data.columns,
        },
    )

# genre from my_sql
@asset(
    description="Load table 'genre' from MySQL database as polars DataFrame, and save to minIO",
    io_manager_key="minio_io_manager",
    required_resource_keys={"mysql_io_manager"},
    key_prefix=["bronze", "goodreads"],
    compute_kind=COMPUTE_KIND,
    group_name=LAYER,
)
def bronze_genre(context) -> Output[pl.DataFrame]:
    query = "SELECT * FROM genre;"
    df_data = context.resources.mysql_io_manager.extract_data(query)
    context.log.info(f"Table extracted with shape: {df_data.shape}")

    return Output(
        value=df_data,
        metadata={
            "table": "genre",
            "row_count": df_data.shape[0],
            "column_count": df_data.shape[1],
            "columns": df_data.columns,
        },
    )

@contextmanager
def connect_minio(config):
    client = Minio(
        endpoint=config.get("endpoint_url"),
        access_key=config.get("minio_access_key"),
        secret_key=config.get("minio_secret_key"),
        secure=False,
    )

    try:
        yield client
    except Exception as e:
        raise e

# Make bucket if not exists
def make_bucket(client: Minio, bucket_name):
    found = client.bucket_exists(bucket_name)
    if not found:
        client.make_bucket(bucket_name)
    else:
        print(f"Bucket {bucket_name} already exists.")

    def _get_path(self, context: Union[InputContext, OutputContext]):
        # E.g context.asset_key.path: ['bronze', 'goodreads', 'book']
        layer, schema, table = context.asset_key.path
        # NOTE: E.g: bronze/goodreads/book
        key = "/".join([layer, schema, table.replace(f"{layer}_", "")])
        # E.g /tmp/file_bronze_goodreads_book_20210101000000.parquet
        tmp_file_path = "/tmp/file_{}_{}.parquet".format(
            "_".join(context.asset_key.path), datetime.today().strftime("%Y%m%d%H%M%S")
        )  # Partition by year

        if context.has_partition_key:
            # E.g partition_str: book_2021
            partition_str = str(table) + "_" + context.asset_partition_key
            # E.g key_name: bronze/goodreads/book/book_2021.parquet
            # tmp_file_path: /tmp/file_bronze_goodreads_book_20210101000000.parquet
            return os.path.join(key, f"{partition_str}.parquet"), tmp_file_path
        else:
            # E.g key_name: bronze/goodreads/book.parquet
            return f"{key}.parquet", tmp_file_path

    def handle_output(self, context: "OutputContext", obj: pl.DataFrame):
        key_name, tmp_file_path = self._get_path(context)

        # Convert from polars DataFrame to parquet format
        obj.write_parquet(tmp_file_path)

        # Upload file to minIO
        try:
            bucket_name = self._config.get("bucket")
            with connect_minio(self._config) as client:
                # Make bucket if not exist
                make_bucket(client, bucket_name)

                # Upload file to minIO
                # E.g bucket_name: lakehouse,
                # key_name: bronze/goodreads/book/book_2021.parquet,
                # tmp_file_path: /tmp/file_bronze_goodreads_book_20210101000000.parquet
                client.fput_object(bucket_name, key_name, tmp_file_path)
                context.log.info(
                    f"(MinIO handle_output) Number of rows and columns: {obj.shape}"
                )
                context.add_output_metadata({"path": key_name, "tmp": tmp_file_path})

                # Clean up tmp file
                os.remove(tmp_file_path)
        except Exception as e:
            raise e

    def load_input(self, context: "InputContext") -> pl.DataFrame:
        """
        Prepares input for downstream asset,
        and downloads parquet file from minIO and converts to polars DataFrame
        """

        bucket_name = self._config.get("bucket")
        key_name, tmp_file_path = self._get_path(context)

        try:
            with connect_minio(self._config) as client:
                # Make bucket if not exist
                make_bucket(client=client, bucket_name=bucket_name)

                # E.g bucket_name: lakehouse,
                # key_name: bronze/goodreads/book/book_2021.parquet,
                # tmp_file_path: /tmp/file_bronze_goodreads_book_20210101000000.parquet
                context.log.info(f"(MinIO load_input) from key_name: {key_name}")
                client.fget_object(bucket_name, key_name, tmp_file_path)
                df_data = pl.read_parquet(tmp_file_path)
                context.log.info(
                    f"(MinIO load_input) Got polars dataframe with shape: {df_data.shape}"
                )

                return df_data
        except Exception as e:
            raise e

# Silver cleaned book
@asset(
    description="Load book table from bronze layer in minIO, into a Spark dataframe, then clean data",
    partitions_def=YEARLY,
    ins={
        "bronze_book": AssetIn(
            key_prefix=["bronze", "goodreads"],
        ),
    },
    io_manager_key="spark_io_manager",
    key_prefix=["silver", "goodreads"],
    compute_kind=COMPUTE_KIND,
    group_name=LAYER,
)
def silver_cleaned_book(context, bronze_book: pl.DataFrame):
    """
    Load book table from bronze layer in minIO, into a Spark dataframe, then clean data
    """

    config = {
        "endpoint_url": os.getenv("MINIO_ENDPOINT"),
        "minio_access_key": os.getenv("MINIO_ACCESS_KEY"),
        "minio_secret_key": os.getenv("MINIO_SECRET_KEY"),
    }

    context.log.debug("Start creating spark session")

    with get_spark_session(config, str(context.run.run_id).split("-")[0]) as spark:
        # Convert bronze_book from polars DataFrame to Spark DataFrame
        pandas_df = bronze_book.to_pandas()
        context.log.debug(
            f"Converted to pandas DataFrame with shape: {pandas_df.shape}"
        )

        spark_df = spark.createDataFrame(pandas_df)
        spark_df.cache()
        context.log.info("Got Spark DataFrame")

        # Dedupe books
        spark_df = spark_df.dropDuplicates()
        # Drop rows with null value in column 'Name'
        spark_df = spark_df.na.drop(subset=["Name"])
        # Drop rows with null values in column ISBN
        spark_df = spark_df.na.drop(subset=["isbn"])
        # Drop rows will null values in column 'Language'
        spark_df = spark_df.na.drop(subset=["Language"])
        # Drop rows with value '--' in column 'Language'
        spark_df = spark_df.filter(spark_df.Language != "--")
        # Drop rows with value > 350 in column 'PagesNumber'
        spark_df = spark_df.filter(spark_df.PagesNumber <= 350)
        # Drop column 'CountsOfReview' (overlap with 'RatingDistTotal')
        spark_df = spark_df.drop("CountsOfReview")
        # Choose rows with 'PublishYear' from 1900 to datetime.today().year
        spark_df = spark_df.filter(
            (spark_df.PublishYear >= 1975)
            & (spark_df.PublishYear <= datetime.today().year)
        )
        # Update value of column 'RatingDist...', splitting by ':' and take the second value
        spark_df = spark_df.withColumn(
            "RatingDist5", split_take_second(col("RatingDist5"))
        )
        spark_df = spark_df.withColumn(
            "RatingDist4", split_take_second(col("RatingDist4"))
        )
        spark_df = spark_df.withColumn(
            "RatingDist3", split_take_second(col("RatingDist3"))
        )
        spark_df = spark_df.withColumn(
            "RatingDist2", split_take_second(col("RatingDist2"))
        )
        spark_df = spark_df.withColumn(
            "RatingDist1", split_take_second(col("RatingDist1"))
        )
        spark_df = spark_df.withColumn(
            "RatingDistTotal", split_take_second(col("RatingDistTotal"))
        )
        # Cast column 'RatingDist...' to Interger
        spark_df = spark_df.withColumn(
            "RatingDist5", spark_df.RatingDist5.cast("Integer")
        )
        spark_df = spark_df.withColumn(
            "RatingDist4", spark_df.RatingDist4.cast("Integer")
        )
        spark_df = spark_df.withColumn(
            "RatingDist3", spark_df.RatingDist3.cast("Integer")
        )
        spark_df = spark_df.withColumn(
            "RatingDist2", spark_df.RatingDist2.cast("Integer")
        )
        spark_df = spark_df.withColumn(
            "RatingDist1", spark_df.RatingDist1.cast("Integer")
        )
        spark_df = spark_df.withColumn(
            "RatingDistTotal", spark_df.RatingDistTotal.cast("Integer")
        )
        # Change column name 'Count of text reviews' to 'CountOfTextReviews'
        spark_df = spark_df.withColumnRenamed(
            "Count of text reviews", "CountOfTextReviews"
        )
        # Change value of column 'Language' from ['en-US', 'en-GB', 'en-CA'],  to 'eng', from 'nl' to 'nld'
        spark_df = spark_df.withColumn(
            "Language", regexp_replace("Language", "en-US", "eng")
        )
        spark_df = spark_df.withColumn(
            "Language", regexp_replace("Language", "en-GB", "eng")
        )
        spark_df = spark_df.withColumn(
            "Language", regexp_replace("Language", "en-CA", "eng")
        )
        spark_df = spark_df.withColumn(
            "Language", regexp_replace("Language", "nl", "nld")
        )

        spark_df.unpersist()

        return Output(
            value=spark_df,
            metadata={
                "table": "silver_cleaned_book",
                "row_count": spark_df.count(),
                "column_count": len(spark_df.columns),
                "columns": spark_df.columns,
            },
        )

# Silver collected book
@asset(
    description="Collect more infomation about cleaned books, such as authors, number of pages",
    partitions_def=YEARLY,
    ins={
        "silver_cleaned_book": AssetIn(
            key_prefix=["silver", "goodreads"],
            metadata={"full_load": False},
        ),
    },
    io_manager_key="spark_io_manager",
    key_prefix=["silver", "goodreads"],
    compute_kind="OpenLibrary API",
    group_name=LAYER,
)
def silver_collected_book(context, silver_cleaned_book: DataFrame) -> Output[DataFrame]:
    """
    Collect more infomation about cleaned books
    - Authors: if missing
    - Number of pages: if missing
    """

    spark_df = silver_cleaned_book
    context.log.debug("Caching spark_df ...")
    spark_df.cache()

    context.log.info("Starting filling missing data ...")
    null_authors_df = spark_df.filter(
        (spark_df.Authors.isNull()) | (spark_df.Authors == "")
    )
    null_pages_number_df = spark_df.filter((spark_df.PagesNumber.isNull()))

    count = 0
    for row in null_authors_df.select("ISBN").collect():
        isbn = row[0]
        context.log.debug(f"Got isbn: {isbn}")
        if isbn is not None:
            # Get request from OpenLibrary API
            req = requests.get(
                f"https://openlibrary.org/api/books?bibkeys=ISBN:{isbn}&format=json&jscmd=data"
            )
            json = req.json()
            if len(json.keys()) > 0:
                context.log.debug("Got json with data")
                # Check if spark_df with column 'ISBN' = isbn has missing value in column 'Authors'
                row_from_df = spark_df.filter(spark_df.ISBN == isbn).collect()[0]
                if row_from_df.Authors is None or row_from_df.Authors is "":
                    context.log.debug("Authors is missing, start filling ...")
                    # Take the first author
                    author = json.get(f"ISBN:{isbn}" or {}).get("authors" or [])
                    author = author[0].get("name") if len(author) > 0 else None
                    if author:
                        count += 1
                        # Update spark_df with column 'ISBN' = isbn and column 'Authors' = author
                        spark_df = spark_df.withColumn(
                            "Authors",
                            when(
                                (spark_df.ISBN == isbn)
                                & (
                                    (spark_df.Authors.isNull())
                                    | (spark_df.Authors == "")
                                ),
                                author,
                            ).otherwise(spark_df.Authors),
                        )
    context.log.info(f"Filled in {count} authors")

    count = 0
    for row in null_pages_number_df.select("ISBN").collect():
        isbn = row[0]
        context.log.debug(f"Got isbn: {isbn}")
        if isbn is not None:
            # Get request from OpenLibrary API
            req = requests.get(
                f"https://openlibrary.org/api/books?bibkeys=ISBN:{isbn}&format=json&jscmd=data"
            )
            json = req.json()
            if len(json.keys()) > 0:
                context.log.debug("Got json with real data")
                # Check if spark_df with column 'ISBN' = isbn has missing value in column 'Authors'
                row_from_df = spark_df.filter(spark_df.ISBN == isbn).collect()[0]
                # Check if spark_df with column 'ISBN' = isbn has missing value in column 'PagesNumber'
                if row_from_df.PagesNumber is None or row_from_df.PagesNumber == 0:
                    context.log.debug("PagesNumber is missing, start filling ...")
                    # Take the number of pages
                    pages_number = json.get(f"ISBN:{isbn}" or {}).get("number_of_pages")
                    if pages_number:
                        count += 1
                        # Update spark_df with column 'ISBN' = isbn and column 'PagesNumber' = pages_number
                        spark_df = spark_df.withColumn(
                            "PagesNumber",
                            when(
                                (spark_df.ISBN == isbn)
                                & (spark_df.PagesNumber.isNull()),
                                pages_number,
                            ).otherwise(spark_df.PagesNumber),
                        )
    context.log.info(f"Filled in {count} pages numbers")

    spark_df.unpersist()

    return Output(
        value=spark_df,
        metadata={
            "table": "silver_collected_book",
            "row_count": spark_df.count(),
            "column_count": len(spark_df.columns),
            "columns": spark_df.columns,
        },
    )

@contextmanager
def connect_psql(config):
    try:
        yield psycopg2.connect(
            host=config["host"],
            port=config["port"],
            database=config["database"],
            user=config["user"],
            password=config["password"],
        )
    except (Exception) as e:
        print(f"Error while connecting to PostgreSQL: {e}")

    def handle_output(self, context: "OutputContext", obj: pl.DataFrame):
        # E.g context.asset_key.path = ['warehouse', 'gold', 'book_genre']
        schema = context.asset_key.path[-2]
        # NOTE: Replace pattern is 'warehouse', not general
        table = str(context.asset_key.path[-1]).replace("warehouse_", "")
        context.log.debug(f"Schema: {schema}, Table: {table}")
        tmp_tbl = f"{table}_tmp_{datetime.now().strftime('%Y_%m_%d')}"
        try:
            with connect_psql(self._config) as conn:
                context.log.debug(f"Connected to PostgreSQL: {conn}")
                primary_keys = (context.metadata or {}).get("primary_keys", [])
                context.log.debug(f"Primary keys: {primary_keys}")

                with conn.cursor() as cursor:
                    context.log.debug(f"Cursor info: {cursor}")
                    cursor.execute("SELECT version()")
                    context.log.info(f"PostgreSQL version: {cursor.fetchone()}")
                    # Create temp file
                    cursor.execute(
                        f"CREATE TEMP TABLE IF NOT EXISTS {tmp_tbl} (LIKE {schema}.{table})"
                    )
                    cursor.execute(f"SELECT COUNT(*) FROM {tmp_tbl}")
                    context.log.debug(
                        f"Log for creating temp table: {cursor.fetchone()}"
                    )
                    # Create sql identifiers for the column names
                    # Do this to safely insert into a sql query
                    columns = sql.SQL(",").join(
                        sql.Identifier(name.lower()) for name in obj.columns
                    )
                    # Create a placeholder for the values. These will be filled later
                    values = sql.SQL(",").join(sql.Placeholder() for _ in obj.columns)
                    # Create the insert query
                    context.log.debug("Inserting data into temp table")
                    insert_query = sql.SQL("INSERT INTO {} ({}) VALUES({});").format(
                        sql.Identifier(tmp_tbl), columns, values
                    )
                    # Execute the insert query
                    psycopg2.extras.execute_batch(cursor, insert_query, obj.rows())
                    conn.commit()

                    # Check data inserted
                    context.log.debug("Checking data inserted")
                    cursor.execute(f"SELECT COUNT(*) FROM {tmp_tbl};")
                    context.log.info(f"Number of rows inserted: {cursor.fetchone()}")
                    # Upsert data
                    if len(primary_keys) > 0:
                        context.log.debug("Table has primary keys, upserting data")
                        conditions = " AND ".join(
                            [
                                f""" {schema}.{table}."{k}" = {tmp_tbl}."{k}" """
                                for k in primary_keys
                            ]
                        )
                        command = f"""
                            BEGIN TRANSACTION;
                            DELETE FROM {schema}.{table}
                            USING {tmp_tbl}
                            WHERE {conditions};

                            INSERT INTO {schema}.{table}
                            SELECT * FROM {tmp_tbl};

                            END TRANSACTION;
                        """
                    else:
                        context.log.debug("Table has no primary keys, replacing data")
                        command = f"""
                            BEGIN TRANSACTION;
                            DELETE FROM {schema}.{table};

                            INSERT INTO {schema}.{table}
                            SELECT * FROM {tmp_tbl};

                            END TRANSACTION;
                        """

                    # context.log.debug(f"Command: {command}")
                    context.log.debug(f"Upserting data into {schema}.{table}")
                    cursor.execute(command)
                    context.log.debug(f"{cursor.statusmessage}")
                    conn.commit()
        except (Exception) as e:
            print(f"Error while handling output to PostgreSQL: {e}")

        try:
            with connect_psql(self._config) as conn:
                with conn.cursor() as cursor:
                    context.log.debug(f"{cursor.fetchone()}")
                    cursor.execute(f"SELECT COUNT(*) FROM {schema}.{table};")
                    context.log.info(
                        f"Number of rows upserted in {schema}.{table}: {cursor.fetchone()}"
                    )

                    # Drop temp table
                    cursor.execute(f"DROP TABLE {tmp_tbl}")
                    conn.commit()
        except (Exception) as e:
            print(f"Error while testing handle_output to PostgreSQL: {e}")

select
  isbn,
  name
from {{ source('gold', 'book_with_info') }}

select
  isbn,
  name
from {{ source('gold', 'book_with_info') }}
right join {{ source('recommendations', 'book_download_link') }} using (isbn)
where link is not null

with tmp_avg_rating as (
  select
    isbn,
    rating
  from {{ source('gold', 'book_with_rating') }}
),
tmp_download_link as (
  select
    isbn,
    case
      when link is null then 0
      else 1
    end as hasdownloadlink,
    rating
  from {{ source('recommendations', 'book_download_link') }}
  right join tmp_avg_rating using (isbn)
)

select *
from tmp_download_link

make down