<jupyter_start><jupyter_text>Data Engineering Capstone Project Project SummaryThe Capstone project analyses and process data from Brazilian stock market since 2013 until now. Also,the Brazilian economic data from [World Bank](https://data.worldbank.org/).With those data in mind, it is possible to evaluate and take insights, relate country data with stock market.Basically to answer questions like:* Does the stock market influence real economy?* Is the population's education rate related to the increase in GDP?* The raising of company in stock market help us somehow?The project follows the follow steps:* Step 1: Scope the Project and Gather Data* Step 2: Explore and Assess the Data* Step 3: Define the Data Model* Step 4: Run ETL to Model the Data* Step 5: Complete Project Write Up<jupyter_code>!pip install bovespa
!pip install pyspark
!pip install wbgapi
import os

import bovespa
import etl_functions
import wbgapi as wb
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, DateType, DoubleType, MapType
from pyspark.sql import types as T
from pyspark.sql import functions as F<jupyter_output><empty_output><jupyter_text>Step 1: Scope the Project and Gather Data ScopeThe Capstone project analyses and process data from Brazilian stock market since 2013 until now. Also,the Brazilian economic data from [World Bank](https://data.worldbank.org/). Describe and Gather DataDescribe the data sets you're using. Where did it come from? What type of information is included?Bovespa is the main Brazilian stock Exchange. Here is listed all trading data since 2013.It includes the day of the trading, company name, stock id, stock data, such as: price open, price close, price high, price low.World Bank collects all data from all country in the world, since 1960.It has information about environment, education, GDP, economy, social.http://www.b3.com.br/en_us/market-data-and-indices/data-services/market-data/historical-data/equities/historical-quotes/https://api.worldbank.org/v2/en/country/BRA?downloadformat=csvResult will be written in disk, to make it simpler.<jupyter_code>os.environ['JAVA_HOME'] = '/home/charl3ff/.sdkman/candidates/java/current'
spark = SparkSession.builder.enableHiveSupport()\
    .appName("Capstone")\
    .getOrCreate()<jupyter_output><empty_output><jupyter_text>ETL stock market<jupyter_code># Load data trading data from bovespa

input_data = "sample_data/"
trading_file = input_data + "COTAHIST*.txt"

df_data = spark.read.text(trading_file)
schema = StructType([
    StructField('date', DateType(), True),
    StructField('year', IntegerType(), True),
    StructField('month', IntegerType(), True),
    StructField('day', IntegerType(), True),
    StructField('money_volume', T.DoubleType(), True),
    StructField('volume', T.IntegerType(), True),
    StructField('stock_code', StringType(), True),
    StructField('company_name', StringType(), True),
    StructField('price_open', DoubleType(), True),
    StructField('price_close', DoubleType(), True),
    StructField('price_mean', DoubleType(), True),
    StructField('price_high', DoubleType(), True),
    StructField('price_low', DoubleType(), True),
    StructField('variation', DoubleType(), True)
])
# Parse data
@F.udf(returnType=schema)
def record_to_dict(row):
    """
    Transform string into bovespa.Record
    :param row: (string) position string from bovespa.
    :return: parsed Record
    """
    try:
        record = bovespa.Record(row)
    except:
        return None
    return {
        'date': record.date, 'year': record.date.year,
        'month': record.date.month, 'day': record.date.day,
        'money_volume': record.volume,
        'volume': record.quantity,
        'stock_code': record.stock_code, 'company_name': record.company_name,
        'price_open': record.price_open, 'price_close': record.price_close,
        'price_mean': record.price_mean, 'price_high': record.price_high,
        'price_low': record.price_low
    }
dict_df = df_data.select(record_to_dict('value').alias('dict'))
trading_df = dict_df.select("dict.*", "*").drop('dict')
trading_df =trading_df.withColumn('variation', (trading_df['price_close'] - trading_df['price_open']) / 100)
trading_df.printSchema()
trading_df = trading_df.repartition("stock_code")
trading_df.orderBy(F.col('variation'), ascending=False).show(10)
trading_df.orderBy(F.col('money_volume'), ascending=False).show(10)
trading_df.show(10)<jupyter_output>+----------+----+-----+---+------------+--------+----------+------------+----------+-----------+----------+----------+---------+--------------------+
|      date|year|month|day|money_volume|  volume|stock_code|company_name|price_open|price_close|price_mean|price_high|price_low|           variation|
+----------+----+-----+---+------------+--------+----------+------------+----------+-----------+----------+----------+---------+--------------------+
|2020-01-02|2020|    1|  2|2.80483738E8|13591400|     GGBR4|      GERDAU|     20.13|      20.76|     20.63|     20.79|    20.12|0.006300000000000025|
|2020-01-02|2020|    1|  2|    23011.02|    1751|    NVHO11|FII NOVOHORI|     13.01|      13.02|     13.14|     13.25|     13.0|9.999999999999786E-5|
|2020-01-02|2020|    1|  2|    10306.76|      73|    CGAS3F|      COMGAS|    142.88|      140.3|    141.18|    142.88|    140.0|-0.02579999999999984|
|2020-01-02|2020|    1|  2|     34460.3|   11854|    DMMO3F|       DOMMO|       3.0|       2.92|    [...]<jupyter_text>So, we can take all insights from it.Such as:<jupyter_code># Most traded stocks in terms of money
stocks = trading_df.orderBy('money_volume', ascending=False).show(10)
# Most up variation in a day
trading_df.orderBy('variation', ascending=False).show(10)
# Rows
print('Amount of rows', trading_df.count())<jupyter_output>Amount of rows 1251648<jupyter_text>ETL World BankThe `wbgapi` package is the official Python/R library to interact with the [World Bank open data](https://data.worldbank.org/),it is useful to easily retrieve data, instead of download it ourselves.The purpose is to fetch data from Brazil, then it will be possible to compare the stock market and the real economy.<jupyter_code># Where data comes from
wb.source.info().items[:3]
metric_to_schema = {
    'GC.DOD.TOTL.GD.ZS': 'debt',
    'GC.XPN.TOTL.CN': 'total_expense',
    'NY.GDP.MKTP.KD.ZG': 'gdp_growth',
    'NY.GDP.MKTP.CD': 'gdp',
    'NY.GDP.PCAP.CD': 'gdp_per_capita',
    'SP.POP.TOTL': 'population',
    'SP.DYN.LE00.IN': 'life_expectancy',
    'GC.XPN.TOTL.GD.ZS': 'expense_per_gdp',
    'FI.RES.TOTL.CD': 'total_reserves',
    'SE.ADT.LITR.ZS': 'pop_literacy_rate',
    'SE.XPD.TOTL.GD.ZS': 'expenditure_education_per_gdp',
    'BX.KLT.DINV.CD.WD': 'foreign_investment'
}
wb.series.info().items[:3]
# Create spark dataframe
df_raw = wb.data.DataFrame(metric_to_schema.keys(), economy='BRA', numericTimeKeys=True)
economy_data = df_raw.rename(metric_to_schema, axis=0)
years= list(economy_data.columns)
economy_data = economy_data.transpose()
economy_data['year'] = years
economy_data
economy_df = spark.createDataFrame(economy_data)
economy_df.printSchema()
economy_df.show(10)<jupyter_output>+------------------+--------------+----+-------------+---------------+-------------------+-----------------+----------------+-----------------+-----------------------------+---------------+-----------+----+
|foreign_investment|total_reserves|debt|total_expense|expense_per_gdp|                gdp|       gdp_growth|  gdp_per_capita|pop_literacy_rate|expenditure_education_per_gdp|life_expectancy| population|year|
+------------------+--------------+----+-------------+---------------+-------------------+-----------------+----------------+-----------------+-----------------------------+---------------+-----------+----+
|               NaN|    4.328488E8| NaN|          NaN|            NaN|1.51655699125199E10|              NaN|210.109899384623|              NaN|                          NaN|         54.143|7.2179226E7|1960|
|               NaN|   4.7115615E8| NaN|          NaN|            NaN| 1.5236854859469E10|  10.275911554301| 205.04076826426|              NaN|                          NaN[...]<jupyter_text>Step 2: Explore and Assess the Data Explore the DataIdentify data quality issues, like missing values, duplicate data, etc. Cleaning StepsDocument steps necessary to clean the data<jupyter_code># Not necessary<jupyter_output><empty_output><jupyter_text>Step 3: Define the Data Model 3.1 Conceptual Data ModelMap out the conceptual data model and explain why you chose that model Trading Table| Feature        | Type| ------------- |:-------------:|| date         | date    || year         | integer || month        | integer || day          | integer || money_volume | double  || volume       | integer || stock_code   | string  || company_name | string  || price_open   | double  || price_close  | double  || price_mean   | double  || price_high   | double  || price_low    | double  || variation    | double  | Economy Table| Feature        | Type  || ------------- |:-------------:|| foreign_investment            | double || total_reserves                | double || debt                          | double || total_expense                 | double || expense_per_gdp               | double || gdp                           | double || gdp_growth                    | double || gdp_per_capita                | double || pop_literacy_rate             | double || expenditure_education_per_gdp | double || life_expectancy               | double || population                    | double || year                          | long   | 3.2 Mapping Out Data Pipelines1. Read trading data from bovespa2. Parse it then load into spark3. Process data creating custom columns4. Write it as parquet6. Read trading data from world bank as pandas7. Parse it then load into spark8. Process data creating custom columns9. Write it as parquet  Step 4: Run Pipelines to Model the Data  4.1 Create the data modelBuild the data pipelines to create the data model.Complete flow can be found in `etl.py`<jupyter_code># Write code here

def process_economy_data(spark, output_data):
    """
    ETL Brazilian economy data from world bank
    :param spark: (SparkSession) spark session instance
    :param output_data: (string) output file path
    :return: spark dataframe representing economy table
    """

    economy_df = etl_functions.create_economy_df()
    economy_df = spark.createDataFrame(economy_df)
    return etl_functions.create_economy_table(economy_df, output_data)


def process_trading_data(spark, trading_files, output_data):
    """
    ETL trading data.
    :param spark: (SparkSession) spark session instance
    :param trading_files: (string) input file path
    :param output_data: (string) output file path
    :return: spark dataframe of trading data
    """

    trading_df = spark.read.text(paths=trading_files)

    trading_df = etl_functions.raw_trading_to_spark(trading_df)
    trading_df = etl_functions.trading_columns(trading_df)

    return etl_functions.create_trading_table(trading_df, output_data)

def etl():
    input_data = "sample_data"
    output_data = "sample_data/output"

    trading_files = os.path.join(input_data, "COTAHIST_A*.txt")

    trading_df = process_trading_data(spark, trading_files, output_data)
    economy_df = process_economy_data(spark, output_data)

    etl_functions.quality_check(economy_df, 'economy')
    etl_functions.quality_check(trading_df, 'trading')
    etl_functions.quality_check_column(trading_df, 'stock_code')
    etl_functions.quality_check_column(trading_df, 'date')
    etl_functions.quality_check_column(trading_df, 'volume')
    etl_functions.quality_check_column(economy_df, 'gdp')
    etl_functions.quality_check_column(economy_df, 'year')<jupyter_output><empty_output><jupyter_text>4.2 Data Quality ChecksExplain the data quality checks you'll perform to ensure the pipeline ran as expected. These could include: * Integrity constraints on the relational database (e.g., unique key, data type, etc.) * Unit tests for the scripts to ensure they are doing the right thing * Source/Count checks to ensure completeness Run Quality Checks<jupyter_code># Perform quality checks here
etl_functions.quality_check(economy_df, 'economy')
etl_functions.quality_check(trading_df, 'trading')
etl_functions.quality_check_column(trading_df, 'stock_code')
etl_functions.quality_check_column(trading_df, 'date')
etl_functions.quality_check_column(trading_df, 'volume')
etl_functions.quality_check_column(economy_df, 'gdp')
etl_functions.quality_check_column(economy_df, 'year')<jupyter_output>Data quality check passed for economy with 61 records.
Data quality check passed for trading with 1251648 records.
Data quality check passed for table stock_code with 1251648 records.
Data quality check passed for table date with 1251648 records.
Data quality check passed for table volume with 1251646 records.
Data quality check passed for table gdp with 61 records.
Data quality check passed for table year with 61 records.
{'fraction_non_alphanumeric': 0.1548638727439584, 'fraction_numerical': 0.034184765983481184, 'mean_word_length': 3.1382911392405064, 'pattern_counts': {'":': 0, '<': 29, '<?xml version=': 0, '>': 29, 'https://': 4, 'lorem ipsum': 0, 'www.': 1, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 3, 'word_list_counts': {'cursed_substrings.json': 5, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '8', 'dir': 'jupyter-structured-clean-dedup', 'id': '329286', 'license': 'MIT', 'n_tokens_mistral': 4256, 'n_tokens_neox': 3897, 'n_words': 1046, 'path': 'Capstone.ipynb', 'repo_name': 'barddoo/capstone-udacity-data-engineer', 'size': '13076'}