def scrap_voting_website(self, url, text_to_search, container_name):
       page = requests.get(url)
       soup = BeautifulSoup(page.content, "html.parser")

       results = soup.find(
           id="ctl00_ctl50_g_ef1010ac_1ad2_43f3_b329_d1623b7a1246_ctl00_pnlUpdate")

       job_elements = results.find_all(
           "div", class_="row home_calendar hc-detail")

       for job_element in job_elements:
           date = job_element.find("p", class_="date")
           year = job_element.find("p", class_="time")

           date_fmt = date.text.strip()
           year_fmt = year.text.strip()

           links = job_element.find_all("a", href=True)

           # Iterates for all pdfs and downloads the links to access the site
           for link in links:
               for text in text_to_search:
                   if text.casefold() in link.text.strip().casefold():
                       url = link['href']
                       file_utils = FileUtils()
                       file_utils.download_file_upload_az_storage(
                           url, link.text.strip(), date_fmt, year_fmt, container_name)

    def download_file_upload_az_storage(self, *args):
        
        try:
            filename = f"{args[1]}_{args[2]}.{args[3]}.pdf"
            filename = filename.replace('/', '_').replace(' ', '_').replace('\\', '_').replace('-', '_')

            file = Path(f"polls-pdfs/{args[3]}/{filename}")
        
            print(f"Downloading {filename}.", flush=True)
            response = requests.get(args[0])
            file.write_bytes(response.content)

        except requests.exceptions.HTTPError as errh:
            print ("Http Error:",errh, flush=True)
        except requests.exceptions.ConnectionError as errc:
            print ("Error Connecting:",errc, flush=True)
        except requests.exceptions.Timeout as errt:
            print ("Timeout Error:",errt, flush=True)
        except requests.exceptions.RequestException as err:
            print ("RequestException: General Error",err, flush=True)

        azure_utils = AzureUtils()
        container_name = args[4] + "/"+ args[3]

        print(f"Uploading {filename} to container {container_name}.")
        azure_utils.upload_blob_to_container(container_name, file,  filename)

    def get_url_from_pdf_files(self, destination_folder, url_to_filter, arr_url_initiatives):

        for filename in os.listdir(destination_folder):
            PDFFile = open(f"{destination_folder}/{filename}", 'rb')

            PDF = PyPDF2.PdfFileReader(PDFFile)
            pages = PDF.getNumPages()
            key = '/Annots'
            uri = '/URI'
            ank = '/A'

            for page in range(pages):
                pageSliced = PDF.getPage(page)
                pageObject = pageSliced.getObject()
                if key in pageObject.keys():
                    ann = pageObject[key]
                    for a in ann:
                        u = a.getObject()
                        if uri in u[ank].keys() and url_to_filter.casefold() in u[ank][uri].casefold():
                            if not u[ank][uri] in arr_url_initiatives:
                                arr_url_initiatives.append(u[ank][uri])

        return arr_url_initiatives

I = 1
for url in arr_url_initiatives:
        print(f"{str(I)} || Scraping initiative urls to get voting details.", flush=True)
        arr_voting_data = scrapper_utils.scrap_initiative_website(url, arr_voting_data)
        I += 1

import psycopg2
import os
from dotenv import load_dotenv

load_dotenv()  # take environment variables from .env.

db_name = os.environ.get("POSTGRES_DB")
db_host = os.environ.get("POSTGRES_HOST")
db_user = os.environ.get("POSTGRES_USER")
db_pass = os.environ.get("POSTGRES_PASSWORD")
db_port = os.environ.get("POSTGRES_PORT")
db_schema = os.environ.get("POSTGRES_SCHEMA")

class AZdbUtils:

    def connect(self):
        try:
            conn = psycopg2.connect(database=db_name,
                                    host=db_host,
                                    user=db_user,
                                    password=db_pass,
                                    port=db_port)

        except (Exception, psycopg2.DatabaseError) as error:
            print(error)

        return conn

    def upload_data_db(self, arr_voting_data):

        conn = self.connect()
        cursor = conn.cursor()

        for data in arr_voting_data:
            print('Inserting data for initiative: ' + data.get_title(), flush=True)
            cursor.execute('INSERT INTO staging_data.scraped_data (title, text, status, url, date, voting_favor, voting_against, voting_abstention, voting_absent) VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s)',
                           (data.get_title(), data.get_text(), data.get_status(), data.get_url(), data.get_date(), data.get_voting_favor(), data.get_voting_against(), data.get_voting_abstention(), data.get_voting_absent()))

        conn.commit()

        print('Executing transform function', flush=True)
        cursor.execute('select staging_data.func_transform_staging_data();')
        conn.commit()

        if conn is not None:
            conn.close()
            print('Database connection closed.', flush=True)

CREATE OR REPLACE FUNCTION staging_data.func_transform_staging_data()
 RETURNS integer
 LANGUAGE plpgsql
AS $function$
declare 
    SUCCESS INTEGER;
    _row record;
    r_voting_favor text; 
    r_voting_against text;
    r_voting_abstention text;
    r_voting_away text;
    error_loop bool;
    ret_initiative_id BIGINT;   
    party_vote_iterator integer;

    arr_voting_favor  bigint[]      DEFAULT  ARRAY[]::bigint[];
    arr_voting_against  bigint[]      DEFAULT  ARRAY[]::bigint[];
    arr_voting_abstent  bigint[]      DEFAULT  ARRAY[]::bigint[];
    arr_voting_away  bigint[]      DEFAULT  ARRAY[]::bigint[];
    aux_id_ref_party bigint;

    id_status_favor int;
    id_status_against int;
    id_status_abstent int;
    id_status_away int;
    id_initiative_result int;

begin
    SUCCESS = 1;

    truncate table refined_data.initiative cascade;
    truncate table refined_data.initiative_voting  cascade;

    
    select id_ref_voting_status into id_status_favor from refined_data.ref_voting_status rvs where lower(status) = lower('Favor');
    select id_ref_voting_status into id_status_against from refined_data.ref_voting_status rvs where lower(status) = lower('Contra');
    select id_ref_voting_status into id_status_abstent from refined_data.ref_voting_status rvs where lower(status) = lower('Abstenção');
    select id_ref_voting_status into id_status_away from refined_data.ref_voting_status rvs where lower(status) = lower('Ausente');
    

    for _row in select  REGEXP_REPLACE(voting_favor, '{*}*"*', '', 'gm') AS voting_favor,
                        REGEXP_REPLACE(voting_against, '{*}*"*', '', 'gm') AS voting_against,
                        REGEXP_REPLACE(voting_abstention, '{*}*"*', '', 'gm') AS voting_abstention,
                        REGEXP_REPLACE(voting_absent, '{*}*"*', '', 'gm') AS voting_absent,
                        id_scraped_data , title , "text", status , url, "date"
                        

                from
                    staging_data.scraped_data sd
                where
                    trim(status) <> '' and (voting_favor <> '{}' or voting_against <> '{}' or voting_abstention <> '{}' or voting_absent <> '{}')   
                    and trim(date) <> '' 
                order by id_scraped_data asc
        
            loop    
                 ret_initiative_id = -1;
                 arr_voting_favor  := '{}'; 
                 arr_voting_against  := '{}';
                 arr_voting_abstent  := '{}';
                 arr_voting_away  := '{}';
                   
                   select id_ref_initiative_result into id_initiative_result from refined_data.ref_initiative_result rvs where lower(result) = lower(_row.status);
                   
                    raise notice 'Iterate favor parties';
                    for r_voting_favor in select unnest(string_to_array(_row.voting_favor,',')) 
                        loop
                            if r_voting_favor ~ '^[A-Z-]*$' then
                                raise notice 'Favor party % match', r_voting_favor;
                                select id_ref_party into aux_id_ref_party from refined_data.ref_party rp where lower(name) like lower(r_voting_favor);
                                arr_voting_favor := arr_voting_favor || aux_id_ref_party;
                            end if;
                        end loop;
                       
                       raise notice 'Iterate against parties';
                       for r_voting_against in select unnest(string_to_array(_row.voting_against,',')) 
                        loop
                            if r_voting_against ~ '^[A-Z-]*$' then
                                raise notice 'Against party % match', r_voting_against;
                                select id_ref_party into aux_id_ref_party from refined_data.ref_party rp where lower(name) like lower(r_voting_against);
                                arr_voting_against := arr_voting_against || aux_id_ref_party;
                            end if;
                        end loop;
                       
                       raise notice 'Iterate abstentions parties';
                       for r_voting_abstention in select unnest(string_to_array(_row.voting_abstention,',')) 
                        loop
                            if r_voting_abstention ~ '^[A-Z-]*$' then
                                raise notice 'Absentee party % match ', r_voting_abstention;
                                select id_ref_party into aux_id_ref_party from refined_data.ref_party rp where lower(name) like lower(r_voting_abstention);
                                arr_voting_abstent := arr_voting_abstent || aux_id_ref_party;
                            end if;
                        end loop;
                       
                        raise notice 'Iterar absent parties';
                        for r_voting_away in select unnest(string_to_array(_row.voting_absent,',')) 
                        loop
                            if r_voting_away ~ '^[A-Z-]*$' then
                                raise notice 'Absent party % match', r_voting_away;
                                select id_ref_party into aux_id_ref_party from refined_data.ref_party rp where lower(name) like lower(r_voting_away);
                                arr_voting_away := arr_voting_away || aux_id_ref_party;
                            end if;
                        end loop;
                       
                        raise notice 'Inserting initiative: %.', _row.title;
                        insert into refined_data.initiative (title, "text", url, "date", id_result)
                        values (_row.title, _row.text, _row.url, _row.date::date, id_initiative_result::integer)
                        RETURNING id_initiative INTO ret_initiative_id;
                        
                        raise notice 'Inserting favor votes to iniative.';
                        FOREACH party_vote_iterator IN ARRAY arr_voting_favor 
                        LOOP 
                            insert into refined_data.initiative_voting (id_ref_party, id_ref_voting_status, id_initiative) values (party_vote_iterator, id_status_favor, ret_initiative_id);
                        END LOOP;
                    
                        raise notice 'Inserting against votes to iniative.';
                        FOREACH party_vote_iterator IN ARRAY arr_voting_against 
                        LOOP 
                            insert into refined_data.initiative_voting (id_ref_party, id_ref_voting_status, id_initiative) values (party_vote_iterator, id_status_against, ret_initiative_id);
                        END LOOP;
                    
                        raise notice 'Inserting abstent votes to iniative.';
                        FOREACH party_vote_iterator IN ARRAY arr_voting_abstent 
                        LOOP 
                            insert into refined_data.initiative_voting (id_ref_party, id_ref_voting_status, id_initiative) values (party_vote_iterator, id_status_abstent, ret_initiative_id);
                        END LOOP;
                    
                        raise notice 'Inserting away votes to iniative.';
                        FOREACH party_vote_iterator IN ARRAY arr_voting_away 
                        LOOP 
                            insert into refined_data.initiative_voting (id_ref_party, id_ref_voting_status, id_initiative) values (party_vote_iterator, id_status_away, ret_initiative_id);
                        END LOOP;
                       
            
            end loop;

        truncate table staging_data.scraped_data;

return SUCCESS;
end;

$function$
;

#read_twitter_data_stream.py

import tweepy
from kafka import KafkaProducer
from json import dumps
import os
import psycopg2

topic = os.environ.get("TOPIC_NAME")

api_key = os.environ.get("TWITTER_API_KEY")
api_secret = os.environ.get("TWITTER_API_SECRET")
bearer_token = os.environ.get("TWITTER_BEARER_TOKEN")
access_token = os.environ.get("TWITTER_ACCESS_TOKEN")
access_token_secret = os.environ.get("TWITTER_ACCESS_TOKEN_SECRET")

class TwitterStream(tweepy.StreamingClient):
    # This function gets called when the stream is working
    def on_connect(self):
        print("Connected")

    # This function gets called when a tweet passes the stream
    def on_tweet(self, tweet):

        # Displaying tweet in console
        if tweet.referenced_tweets == None:
            print(f"Sending to Kafka: {tweet.text}")
            try:
                producer.send(topic, tweet.text)
                producer.flush()
            except:
                print("An exception occurred")


db_name = os.environ.get("POSTGRES_DB")
db_host = os.environ.get("POSTGRES_HOST")
db_user = os.environ.get("POSTGRES_USER")
db_pass = os.environ.get("POSTGRES_PASSWORD")
db_port = os.environ.get("POSTGRES_PORT")

conn = psycopg2.connect(database=db_name,
                        host=db_host,
                        user=db_user,
                        password=db_pass,
                        port=db_port)

cursor = conn.cursor()

cursor.execute('SET search_path TO streaming_data')
cursor.execute('SELECT value FROM streaming_data.term_search WHERE active is true')

results = cursor.fetchall()

conn.close()

if len(results) > 0:
    for term in results:
        print(f"Printing search term: {term[0]}")

    producer = KafkaProducer(bootstrap_servers=['kafka:29092'],
                             value_serializer=lambda x:
                             dumps(x, ensure_ascii=False).encode('utf-8'))

    client = tweepy.Client(bearer_token, api_key, api_secret,
                           access_token, access_token_secret)

    auth = tweepy.OAuth1UserHandler(
        api_key, api_secret, access_token, access_token_secret)
    api = tweepy.API(auth)

    # Creating Stream object
    stream = TwitterStream(bearer_token=bearer_token)

    print("Search for previous rules.")
    previousRules = stream.get_rules().data

    if previousRules:
        print("Deleting previous rules.")
        stream.delete_rules(previousRules)

    for term in results:
        stream.add_rules(tweepy.StreamRule(term[0]))

    stream.filter(tweet_fields=["referenced_tweets"])
else:
    print("No search term find in database. Aborting.")

kafka-topics --bootstrap-server kafka:29092 --create --if-not-exists --topic tp-twitter-stream --replication-factor 1 --partitions 1

from pyspark.sql import SparkSession
from pyspark.sql.functions import *
import os

db_name = os.environ.get("POSTGRES_DB")
db_host = os.environ.get("POSTGRES_HOST")
db_user = os.environ.get("POSTGRES_USER")
db_pass = os.environ.get("POSTGRES_PASSWORD")
db_port = os.environ.get("POSTGRES_PORT")
db_schema = os.environ.get("POSTGRES_SCHEMA")

db_target_properties = {"user": db_user, "password": db_pass, "driver": 'org.postgresql.Driver'}
db_connection_string = f"jdbc:postgresql://{db_host}:{db_port}/{db_name}?currentSchema={db_schema}"

def split_words(input_expression):
    return explode(split(input_expression, " "))

def foreach_batch_function(df, id):
    df.write.jdbc(url=db_connection_string,table='word_count',  properties=db_target_properties, mode='overwrite')
    pass

print("## Starting Processing ##")

spark = SparkSession \
    .builder \
    .appName("Twitter Stream Word Count") \
    .master("spark://10.0.0.8:7077")\
    .getOrCreate()

spark.sparkContext.setLogLevel("WARN")

print("## Creating session ##")

kafka_input_df = spark \
    .readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "10.0.0.3:29092") \
    .option("subscribe", "tp-twitter-stream") \
    .option("startingOffsets", "earliest") \
    .load()

print("## Listening to Kafka Topic ##")

# Split the lines into words
splited_words = split_words(kafka_input_df.value)

words_df = kafka_input_df.select(splited_words.alias("word"),kafka_input_df.timestamp.alias('time'))

# Take special chars out
words_df_clean = words_df.withColumn("word", regexp_replace(col("word"), "[^0-9a-zA-Z]+", ""))

counts_df = words_df_clean.withWatermark("time", "1 minutes").groupBy("word").count()
final_df = counts_df.writeStream.outputMode("complete").foreachBatch(foreach_batch_function).start()
final_df.awaitTermination()

FROM python:3
ENV PYTHONUNBUFFERED=1
RUN mkdir -p /app
WORKDIR /app

#Installing python dependencies
RUN pip install --upgrade pip
RUN pip install tweepy
RUN pip install kafka-python
RUN pip install psycopg2

COPY ./streaming-files/read_twitter_data_stream.py .
CMD ["/bin/bash", "-c", "/bin/sleep 60 && python ./read_twitter_data_stream.py"]

version: "2"
services:
  zookeeper:
    container_name: zookeeper
    user: "0:0"
    image: "confluentinc/cp-zookeeper:latest"
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    restart: always
    volumes:
      - ./volume-data/zookeeper/data:/var/lib/zookeeper/data
      - ./volume-data/zookeeper/log:/var/lib/zookeeper/log
    networks:
      containers-network:
        ipv4_address: 10.0.0.2

  kafka:
    container_name: kafka
    user: "0:0"
    image: "confluentinc/cp-kafka:latest"
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"
      KAFKA_ADVERTISED_LISTENERS: "PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092"
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: "PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT"
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
    restart: always
    volumes:
      - ./volume-data/kafka/data:/var/lib/kafka/data
    networks:
      containers-network:
        ipv4_address: 10.0.0.3

  control-center:
    container_name: control-center
    image: "confluentinc/cp-enterprise-control-center:latest"
    hostname: control-center
    depends_on:
      - zookeeper
      - kafka
    ports:
      - "9021:9021"
    environment:
      CONTROL_CENTER_BOOTSTRAP_SERVERS: "kafka:29092"
      CONTROL_CENTER_ZOOKEEPER_CONNECT: "zookeeper:2181"
      CONTROL_CENTER_REPLICATION_FACTOR: 1
      CONTROL_CENTER_INTERNAL_TOPICS_PARTITIONS: 1
      CONTROL_CENTER_MONITORING_INTERCEPTOR_TOPIC_PARTITIONS: 1
      CONFLUENT_METRICS_TOPIC_REPLICATION: 1
      PORT: 9021
    restart: always
    networks:
      containers-network:
        ipv4_address: 10.0.0.4

  init-kafka:
    container_name: init-kafka
    image: confluentinc/cp-kafka:latest
    depends_on:
      - kafka
    entrypoint: ["/bin/sh", "-c"]
    command: |
      "
      # blocks until kafka is reachable
      kafka-topics --bootstrap-server kafka:29092 --list

      echo -e 'Creating kafka topics'
      kafka-topics --bootstrap-server kafka:29092 --create --if-not-exists --topic tp-twitter-stream --replication-factor 1 --partitions 1

      echo -e 'Successfully created the following topics:'
      kafka-topics --bootstrap-server kafka:29092 --list
      "
    networks:
      containers-network:
        ipv4_address: 10.0.0.5

  python-script:
    container_name: python-script
    build: ./
    image: python:3
    depends_on:
      - zookeeper
      - kafka
      - init-kafka
      - pg-streaming-db
    environment:
      - PYTHONUNBUFFERED=1
    networks:
      containers-network:
        ipv4_address: 10.0.0.6
    env_file:
      - .env

  pg-streaming-db:
    container_name: pg-streaming-db
    image: postgres:14.5-alpine
    env_file:
      - .env
    volumes:
      - ./volume-data/pg-streaming-db/data:/var/lib/postgresql/data
      - ./sql-scripts/:/docker-entrypoint-initdb.d/
    networks:
      containers-network:
        ipv4_address: 10.0.0.7
    ports:
      - 40000:5432
    restart: always

  spark:
    container_name: spark
    image: docker.io/bitnami/spark:3.3
    environment:
      - SPARK_MODE=master
      - SPARK_RPC_AUTHENTICATION_ENABLED=no
      - SPARK_RPC_ENCRYPTION_ENABLED=no
      - SPARK_LOCAL_STORAGE_ENCRYPTION_ENABLED=no
      - SPARK_SSL_ENABLED=no
    ports:
      - "8080:8080"
    networks:
      containers-network:
        ipv4_address: 10.0.0.8
    depends_on:
      - kafka
      - python-script
    env_file:
      - .env
    restart: always

  spark-worker-1:
    container_name: spark-worker-1
    image: docker.io/bitnami/spark:3.3
    environment:
      - SPARK_MODE=worker
      - SPARK_MASTER_URL=spark://spark:7077
      - SPARK_WORKER_MEMORY=2G
      - SPARK_WORKER_CORES=1
      - SPARK_RPC_AUTHENTICATION_ENABLED=no
      - SPARK_RPC_ENCRYPTION_ENABLED=no
      - SPARK_LOCAL_STORAGE_ENCRYPTION_ENABLED=no
      - SPARK_SSL_ENABLED=no
    networks:
      containers-network:
        ipv4_address: 10.0.0.9
    depends_on:
      - spark
    #command: "spark-submit --master spark://spark:7077 --packages org.apache.spark:spark-sql-kafka-0-10_2.12:3.3.0 /opt/bitnami/spark/spark-scripts/spark_stream_process.py"
    volumes:
      - ./spark-scripts/:/opt/bitnami/spark/spark-scripts
    restart: always

  spark-worker-2:
    container_name: spark-worker-2
    image: docker.io/bitnami/spark:3.3
    environment:
      - SPARK_MODE=worker
      - SPARK_MASTER_URL=spark://spark:7077
      - SPARK_WORKER_MEMORY=2G
      - SPARK_WORKER_CORES=1
      - SPARK_RPC_AUTHENTICATION_ENABLED=no
      - SPARK_RPC_ENCRYPTION_ENABLED=no
      - SPARK_LOCAL_STORAGE_ENCRYPTION_ENABLED=no
      - SPARK_SSL_ENABLED=no
    networks:
      containers-network:
        ipv4_address: 10.0.0.10
    depends_on:
      - spark
    restart: always

  metabase-app:
    container_name: metabase-app
    image: metabase/metabase
    ports:
      - 3000:3000
    environment:
      MB_DB_TYPE: postgres
      MB_DB_DBNAME: metabase
      MB_DB_PORT: ${POSTGRES_PORT}
      MB_DB_USER: ${POSTGRES_USER}
      MB_DB_PASS: ${POSTGRES_PASSWORD}
      MB_DB_HOST: ${POSTGRES_HOST}
    depends_on:
      - pg-streaming-db
    links:
      - pg-streaming-db
    env_file:
      - .env
    networks:
      containers-network:
        ipv4_address: 10.0.0.11
    restart: always

networks:
  containers-network:
    driver: bridge
    ipam:
      config:
        - subnet: 10.0.0.0/16
          gateway: 10.0.0.1

docker compose build --no-cache

docker compose up -d

docker cp ./spark-scripts/spark_stream_process.py spark:/opt/bitnami/spark/spark_stream_process.py

docker exec spark /opt/bitnami/spark/bin/spark-submit --packages org.postgresql:postgresql:42.5.0,org.apache.spark:spark-sql-kafka-0-10_2.12:3.3.0 /opt/bitnami/spark/spark_stream_process.py

delete from word_count wc where length(word) > 20 or length(word) < 2;

def download_data(directory):
    logging.info(f"Starting ingestion - Searching for files in directory: {directory}")

    try:
      logging.info("Instancianting KaggleAPI.")
      api = KaggleApi()

      logging.info("Authenticating KaggleAPI.")
      api.authenticate()

      logging.info("Downloading files ... ")
      api.dataset_download_files('rohanrao/formula-1-world-championship-1950-2020',  path=f"{directory}")

      list_dir = os.listdir(f"{directory}")
      total_downloaded_files = len(list_dir)

      logging.info(f"Downloaded {total_downloaded_files} files.")

      if(total_downloaded_files == 0):
        raise Exception("Aborting. No files downloaded") 

      logging.info("Extracting files ... ")
      with zipfile.ZipFile(f"{directory}/formula-1-world-championship-1950-2020.zip", 'r') as zipref:
          zipref.extractall(f"{directory}")

      for item in list_dir:
        if item.endswith(".zip"):
          logging.info(f"Removing file: {item}.")
          os.remove(os.path.join(directory, item))

    except Exception as error:
       logging.error(f"Error loading data. Error: {error}")
       raise
     

def convert_csv_to_parquet(directory):
    logging.info("Starting conversion from CSV to parquet files.")

    try:
        for root, dirs, files in os.walk(f"{directory}"):
            for file in files:
                p=os.path.join(root,file)
                full_path = os.path.abspath(p)

                if full_path.endswith('.csv'):
                    logging.info(f"Reading file {full_path} in CSV format")        
                    table = pyarrowcsv.read_csv(full_path)
                    logging.info(f"Writing file {full_path} to parquet format")        
                    pyarrowpqt.write_table(table, full_path.replace('.csv', '.parquet'))
    
    except Exception as error:
       logging.error(f"Error converting CSV data to parquet. Error: {error}")
       raise

def upload_data_gcs(directory, bucket_id):
    logging.info(f"Starting upload to GCS Bucket: {bucket_id} from directory: {directory} ")

    try:
        for root, dirs, files in os.walk(f"{directory}"):
            for file in files:
                p=os.path.join(root,file)
                parquet_file_full_path = os.path.abspath(p)

                if file.endswith('.parquet'):
                    logging.info(f"Authenticating in GCS to send parquet_file: {file} ")
                    client = storage.Client()
                    bucket = client.bucket(bucket_id)

                    logging.info("Creating blob.")
                    blob = bucket.blob(f"f1-data/raw-files/{date.today().year}/{file}")

                    logging.info(f"Uploading file from location: {parquet_file_full_path} ")
                    blob.upload_from_filename(parquet_file_full_path)
    
    except Exception as error:
       logging.error(f"Error uploading data. Error: {error}")
       raise

def load_staging_files(directory, project_id, dataset_name, bucket_uri):
    logging.info(f"Starting loading files for staging in BigQuery.")

    try:
        client = bigquery.Client()

        job_config = bigquery.LoadJobConfig(
            source_format=bigquery.SourceFormat.PARQUET,
        )

        for root, dirs, files in os.walk(f"{directory}"):
            for file in files:
                if file.endswith('.parquet'):
                    bucket_file_location = str(bucket_uri) + str(date.today().year) + '/' + str(file)
                    logging.info(f"Searching file {bucket_file_location} from location: {bucket_uri}")

                    table_name = f"stg_{Path(f'{file}').stem}"
                    table_id = f"{project_id}.{dataset_name}.{table_name}"
                    logging.info(f"Uploading data from table: {table_id}")

                    load_job = client.load_table_from_uri(
                        bucket_file_location, table_id, job_config=job_config
                    )  

                    load_job.result()  

                    destination_table = client.get_table(table_id)
                    logging.info(f"Loaded: {destination_table.num_rows} rows.")
        
        logging.info(f"Removing all files from {directory} after uploading them to BigQuery.")
        for f in os.listdir(directory):
            os.remove(os.path.join(directory, f))

    except Exception as error:
       logging.error(f"Error loading files for stating dataset. Error: {error}")
       raise

{{ config(materialized='table', alias='driver_winning_rate') }}
with driver as (
    select forename, surname, driverId from {{ source('src-bq-de-f1-project', 'stg_drivers') }}
),
results as (
    select driverId, points, raceId  from {{ source('src-bq-de-f1-project', 'stg_results') }}
),
races as (
      select year, raceId  from {{ source('src-bq-de-f1-project', 'stg_races') }}
),

final as (

    select round(sum(q2.number_of_points)/sum(q1.number_of_participations),1) as winning_rate, q1.driver_name, q1.race_year
    from (
        select concat(driver.forename, ' ', driver.surname) as driver_name, count( distinct results.raceId) as number_of_participations, races.year as race_year, races.raceId as raceId
        from results
        inner join driver on driver.driverId = results.driverId
        inner join races on races.raceId = results.raceId
        group by driver_name, race_year, races.raceId
    ) q1
    left join 
    (select concat(driver.forename, ' ', driver.surname) as driver_name, sum(points)  as number_of_points, races.year as race_year , races.raceId as raceId
        from  results
        inner join driver on driver.driverId = results.driverId
        inner join races on races.raceId = results.raceId
        group by driver_name, race_year, races.raceId
    ) q2 
    on q1.driver_name = q2.driver_name and q1.raceId = q2.raceId and q1.race_year = q2.race_year
    where q2.number_of_points > 0 
    group by q1.driver_name, q1.race_year
)

select * from final

FROM apache/airflow:2.3.3

ENV AIRFLOW_HOME=/opt/airflow

USER ${AIRFLOW_UID}
#Installing python dependencies
RUN pip install --upgrade pip
RUN pip install kaggle
RUN pip install pyarrow
RUN pip install dbt-bigquery

#Making directories for airflow and kaggle.
RUN mkdir -p /opt/airflow/env_files/kaggle/
RUN mkdir -p /opt/airflow/env_files/google/credentials
RUN mkdir -p /opt/airflow/temp/download/kaggle/

#Copy files to previous created folders
COPY .google/credentials/google_credentials.json /opt/airflow/env_files/google/credentials/
COPY .kaggle/kaggle.json /opt/airflow/env_files/kaggle/

#Set permissions
USER root
RUN chown -R airflow /opt/airflow/env_files/
RUN chown -R airflow /opt/airflow/temp/download/kaggle/
RUN chmod 770 /opt/airflow/temp/download/kaggle/

RUN mkdir -p /opt/dbt/

#Copy dbt files 
WORKDIR /opt/dbt/
COPY dbt/dbt_project.yml .
COPY dbt/profiles.yml .
ADD --chown=airflow:root dbt/analyses/ analyses/
ADD --chown=airflow:root dbt/dbt_packages/ dbt_packages/
ADD --chown=airflow:root dbt/logs logs/
ADD --chown=airflow:root dbt/macros macros/
ADD --chown=airflow:root dbt/models models/
ADD --chown=airflow:root dbt/seeds seeds/
ADD --chown=airflow:root dbt/snapshots snapshots/
ADD --chown=airflow:root dbt/target target/
ADD --chown=airflow:root dbt/tests tests/

#Set permissions
RUN chmod -R 770 /opt/dbt/logs
RUN chmod -R 770 /opt/dbt/target

#GCP Specifics 
#Ref: https://airflow.apache.org/docs/docker-stack/recipes.html
USER 0
SHELL ["/bin/bash", "-o", "pipefail", "-e", "-u", "-x", "-c"]
ARG CLOUD_SDK_VERSION=322.0.0
ENV GCLOUD_HOME=/home/google-cloud-sdk
ENV PATH="${GCLOUD_HOME}/bin/:${PATH}"

RUN DOWNLOAD_URL="https://dl.google.com/dl/cloudsdk/channels/rapid/downloads/google-cloud-sdk-${CLOUD_SDK_VERSION}-linux-x86_64.tar.gz" \
    && TMP_DIR="$(mktemp -d)" \
    && curl -fL "${DOWNLOAD_URL}" --output "${TMP_DIR}/google-cloud-sdk.tar.gz" \
    && mkdir -p "${GCLOUD_HOME}" \
    && tar xzf "${TMP_DIR}/google-cloud-sdk.tar.gz" -C "${GCLOUD_HOME}" --strip-components=1 \
    && "${GCLOUD_HOME}/install.sh" \
       --bash-completion=false \
       --path-update=false \
       --usage-reporting=false \
       --quiet \
    && rm -rf "${TMP_DIR}" \
    && gcloud --version

WORKDIR $AIRFLOW_HOME

USER $AIRFLOW_UID

  postgres-metabase:
    image: postgres:13
    environment:
      POSTGRES_USER: <hidden>
      POSTGRES_PASSWORD: <hidden>
      POSTGRES_DB: <hidden>
    volumes:
      - /opt/docker/postgres-data:/var/lib/postgresql/data
    ports:
      - 5433:5432
    restart: always 

  metabase-app:
    image: metabase/metabase
    restart: always
    ports:
      - 3000:3000
    environment:
      MB_DB_TYPE: postgres
      MB_DB_DBNAME: <hidden>
      MB_DB_PORT: 5432
      MB_DB_USER: <hidden>
      MB_DB_PASS: <hidden>
      MB_DB_HOST: postgres-metabase
    depends_on:
      - postgres-metabase
    links:
     - postgres-metabase