Accessing data using FastAPI and Streamlit

After setting up our infrastructure and data pipeline using DuckDB and Prefect in Part 1 of this series of articles, we can now proceed to accessing and consuming our sensor data using a modern data stack API framework and visualization tool.

For our API we will be using FastAPI — a modern Python framework for developing API. For the visualization, we will learn how to use and run Streamlit — a modern open-source dashboard and visualization tool for building data apps, also using Python.

A. Building the Data API

Like any other Python package, we first need to install FastAPI using pip or any package manager of your choice.

pip install fastapi
pip install "uvicorn[standard]"

A sample FastAPI app is something like below. You can see more at the FastAPI documentation.

from typing import Union

from fastapi import FastAPI

app = FastAPI()


@app.get("/")
def read_root():
    return {"Hello": "World"}


@app.get("/items/{item_id}")
def read_item(item_id: int, q: Union[str, None] = None):
    return {"item_id": item_id, "q": q}

To execute this, save the code above in the file name main.py then execute the command below

uvicorn main:app --reload

INFO:     Uvicorn running on http://127.0.0.1:8000 (Press CTRL+C to quit)
INFO:     Started reloader process [28720]
INFO:     Started server process [28722]
INFO:     Waiting for application startup.
INFO:     Application startup complete.

With that out of the way, we can follow a similar pattern to create a basic API using FastAPI. Below is the basic API I have created to access data in a DuckDB database.

#
# Main.py
#

from typing import Union
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
import io

from  .data  import Database

app = FastAPI()
db = Database()

@app.get("/")
def read_root():
    return {"Hello": "World"}


@app.get("/items/{item_id}")
def read_item(item_id: int, q: Union[str, None] = None):
    return {"item_id": item_id, "q": q}

@app.get("/api/v1/daily_avg", response_class=StreamingResponse)
def get_daily_avg():
    avg_df = db.get_daily_avg()
    avg_df = avg_df.reset_index()
    
    return _to_csv(avg_df,'daily_avg')

@app.get("/api/v1/daily_max", response_class=StreamingResponse)
def get_daily_max():
    max_df = db.get_daily_max()
    max_df = max_df.reset_index()
    
    return _to_csv(max_df,'daily_max')

@app.get("/api/v1/daily_min" ,response_class=StreamingResponse)
def get_daily_min():
    min_df = db.get_daily_min()
    min_df = min_df.reset_index()
    
    return _to_csv(min_df,'daily_min')


@app.get("/api/v1/metadata", response_class=StreamingResponse)
def get_metadata():
    mdf =  db.get_gauge_metadata()
    return _to_csv(mdf,'metadata')

@app.get("/api/v1/sensor_metrics", response_class=StreamingResponse)
def get_sensor_metrics():
    print("api/v1/sensor_metrics")
    gdf = db.get_gauge_metrics()
    gdf = gdf.reset_index()

    return _to_csv(gdf,'metrics')


def _to_csv(df, fname):
    stream = io.StringIO()
    df.to_csv(stream, index=False)
    response = StreamingResponse(
        iter([stream.getvalue()]), media_type="text/csv")
    response.headers["Content-Disposition"] = f"attachment; filename={fname}.csv"

    return response

I also created another file to represent our DuckDB database.

#
# data.py
#
import duckdb

def get_conn(db=':memory:',is_shared=False):
    return duckdb.connect(database=db, read_only=is_shared)

class Database():

    def __init__(self):
        self.conn = get_conn('db/datahack.duckdb')
        self._metrics_df = self.__init_metrics_df(self.conn)
        self._attributes = self.__get_attributes()
      
    def __init_metrics_df(self, conn):
        metrics_df = conn.sql('SELECT * EXCLUDE(measured_day,measured_hour,measured_minute) FROM ingest_gauge_metrics').df()
        metrics_df = metrics_df.set_index("measured_datetime")
        metrics_df = metrics_df.astype('Float64')
        
        conn.close()
        return metrics_df

    def __get_attributes(self):
        conn = get_conn('db/datahack.duckdb')
        attrs_df = conn.sql('SELECT *  FROM gauge_sensors_metadata').df()
        conn.close()
        return attrs_df #.to_csv(index=False)

    @property
    def metrics_df(self):
        return self._metrics_df
    
    @property
    def attributes(self):
        return self._attributes

    def get_gauge_metrics(self):
        print('Get metrics')
        return self.metrics_df
     
    def get_gauge_metadata(self):
        print('Get metadata')
        return self.attributes
    
    def get_daily_avg(self):
        print('Daily Avg')
        daily_mean = self.metrics_df.resample("D").mean()
        print(daily_mean)

        return daily_mean

    def get_daily_max(self):
        print('Daily Max')
        daily_max = self.metrics_df.resample("D").max()
        print(daily_max)

        return daily_max

    def get_daily_min(self):
        print('Daily Min')
        daily_min = self.metrics_df.resample("D").min()
        print(daily_min)

        return daily_min

To test the API, we simply run this command

uvicorn src.api.main:app --host 0.0.0.0 --proxy-headers --reload

Now, going to this URL http://0.0.0.0:8000/docs will show us the documentation as we will as a testing option as shown below:

To test the metadata endpoint, we just click on the name provided like below and click execute.

B. Displaying the data using Streamlit

Finally, once our API is up, we can invoke them inside a Streamlit app and build some cool visualization.

To be able to run the Streamlit app, we first need to install it like any other Python package:

pip install streamlit

#testing command line
streamlit hello

#testing with a file
streamlit run hello.py

To learn more about Streamlit you can check out its documentation for advanced visualization.

Below is a simple dashboard that I forked from the DataProfessor GitHub repo and then modified for my use case.

#weatherboard_app.py

import streamlit as st
import pandas as pd
import plost
import altair as alt
import math

from dotenv import load_dotenv
from urllib.request import urlopen
import json
import folium
from streamlit_folium import st_folium, folium_static
from streamlit_plotly_events import plotly_events
import plotly.figure_factory as ff
import plotly.express as px
import numpy as np
from datetime import datetime

load_dotenv()

#API_URL = 'http://0.0.0.0:8000'
API_URL = 'http://datahack-api-1:8000/api/v1'
st.set_page_config(layout='wide', initial_sidebar_state='expanded')

dt_now = datetime.now()
dt_day = dt_now.strftime('%A')
_,colT2 = st.columns([3,7])
with colT2:
    st.title('Brisbane Weatherboard')
    st.write(f"### {dt_day}, {dt_now}")
with open('style.css') as f:
    st.markdown(f'<style>{f.read()}</style>', unsafe_allow_html=True)
    
st.sidebar.header('Weatherboard `v.1.0`')

st.sidebar.subheader('Map parameter')
time_hist_color = st.sidebar.selectbox('Display by sensor', ('Rainfall', 'Stream')) 

st.sidebar.subheader('Heat map parameter')
time_hist_color = st.sidebar.selectbox('Color by', ('min_value', 'max_value')) 

#st.sidebar.subheader('Donut chart parameter')
#donut_theta = st.sidebar.selectbox('Select data', ('Active', 'Inactive'))

st.sidebar.subheader('Line chart parameters')
plot_data = st.sidebar.multiselect('Select data', ['avg_value', 'max_value','min_value'], ['avg_value', 'min_value','max_value'])
plot_height = st.sidebar.slider('Specify plot height', 0,2,1)

st.sidebar.markdown('''
---
Created with ❤️ by [Data Professor](https://youtube.com/dataprofessor/).\n\n
                    
Modified with passion by Datamatiks.
''')

# Metrics row 
st.markdown('### Weather Metrics')
col1, col2, col3 = st.columns(3)

#Retrive other weather data from public api
response = urlopen("https://wttr.in/Brisbane?format=j1")
json_data = response.read().decode('utf-8', 'replace')
d = json.loads(json_data)

df_brissy = pd.json_normalize(d['current_condition'])
temp = df_brissy['temp_C'].values[0]
wind = df_brissy['windspeedKmph'].values[0]
humid = df_brissy['humidity'].values[0]
col1.metric("Temperature", f"{temp} °C", "1.2 °C")
col2.metric("Wind", f"{wind} kph", "-8%")
col3.metric("Humidity", f"{humid}%", "4%")

#retrive metadata
metadata = pd.read_csv(
    f'{API_URL}/metadata',
    header=0, 
    names=['sensor_id', 
            'location_id',
            'location_name',
            'sensor_type',
            'unit',
            'latitude',
            'longitude']
)

#retrieve daily metrics
sensors_df = pd.read_csv(f'{API_URL}/sensor_metrics',header=0)
avg_df = pd.read_csv(f'{API_URL}/daily_avg',header=0)
max_df = pd.read_csv(f'{API_URL}/daily_max',header=0)
min_df = pd.read_csv(f'{API_URL}/daily_min',header=0)

new_avg_df = avg_df.melt( id_vars=["measured_datetime"], 
         var_name="sensor_id", 
         value_name="avg_value")

new_max_df = max_df.melt( id_vars=["measured_datetime"], 
         var_name="sensor_id", 
         value_name="max_value")

new_min_df = min_df.melt( id_vars=["measured_datetime"], 
         var_name="sensor_id", 
         value_name="min_value")

#merge all metrics
merge_keys=['sensor_id','measured_datetime']
daily_metrics_df = new_avg_df.merge(new_max_df, on=merge_keys).merge(new_min_df, on=merge_keys).set_index('sensor_id')

#enrich metrics with metadata
joined_df = daily_metrics_df.join(metadata.set_index('sensor_id')).reset_index()


date_now_str = str(dt_now.date()) 
df_filtered = joined_df.loc[(joined_df['measured_datetime'] == date_now_str)]
df_active = df_filtered[df_filtered['avg_value']>0].groupby('sensor_id').apply(len)
df_inactive = df_filtered[df_filtered['avg_value']==0].groupby('sensor_id').apply(len)
df_unknown = df_filtered[df_filtered['avg_value'].isnull() ].groupby('sensor_id').apply(len)

gauge_status_df = pd.DataFrame(
        {'status':['Active','Inactive','Unknown'],                              
         'count': [len(df_active.index),len(df_inactive.index),len(df_unknown.index)]
        }
)

### Map plotly version
st.markdown('### Rainfall and Stream AHD Height Sensors Map')
px.set_mapbox_access_token(os.environ['MAPBOX_TOKEN'] )
df = metadata.__deepcopy__()

df = joined_df.fillna(0)
fig = px.scatter_mapbox(df, lat="latitude", lon="longitude",  hover_name="location_name", hover_data=["sensor_type", "min_value"], color="avg_value", size="max_value",
                color_continuous_scale=px.colors.cyclical.IceFire, size_max=50, zoom=10, width=1500, height=800)
st.plotly_chart(fig)

#Heatmap
c1, c2 = st.columns((7,3))
with c1:
    st.markdown('### Heatmap')
    plost.time_hist(
    data=joined_df,
    date='measured_datetime',
    x_unit='hours',
    y_unit='day',
    color=time_hist_color,
    aggregate='median',
    legend=None,
    height=345,
    use_container_width=True)
with c2:
    st.markdown('### Sensors status')
    plost.donut_chart(
        data=gauge_status_df,
        theta='count',
        color='status',
        legend='bottom', 
        use_container_width=True)

# Row C
st.markdown('### Line chart')
grouped = joined_df.groupby('sensor_type')
for key, group in grouped:
    st.markdown(f"*** {key}")
    #st.write(group)
    df = group.set_index('measured_datetime')
    df = group.astype({'avg_value': 'float','max_value': 'float','min_value': 'float'})
    plost.line_chart(df, 
        x = 'measured_datetime' ,
        y = ('min_value','max_value','avg_value'),
        color='sensor_id',
        height=plot_height, 
        #pan_zoom='minimap'
    )

st.write(joined_df.reset_index().sort_values(by='measured_datetime', ascending=False))

Running the app like this:

streamlit run src/web/weatherboard_app.py

It will then launch this dashboard.

There we have it, we have access to our DuckDB sensor data using FastAPI and then display the data in a Streamlit dashboard. I added some weather information from another API for weather called https://wttr.in/.

Finally, in the last part of this article, we will discuss how we can package all of these technologies together using docker/docker-compose and then deploy them to the Azure cloud. So, watch this space for it.

Happy reading!

UPDATE: GitHub repo: https://github.com/datamatiks/iot-platform-modern-data-stack-poc

Building an IoT Platform Using Modern Data Stack — Part 2 was originally published in Datamatiks on Medium, where people are continuing the conversation by highlighting and responding to this story.

DuckDB + FastAPI + Prefect + Streamlit (DFAPS Stack)

Last August 18–20, 2023 I intended to join Australia’s GovHack 2023(https://govhack.org/) hackathon but because I had other commitments at that time, I had to give it a pass again this year.

My idea was to create something related to IoT since I am working with IoT data on my day job. I know there will be some open datasets for IoT based on past hackathons. After the event, I browsed through the datasets and luckily enough I found an interesting dataset. Brisbane City Council uploaded the telemetry sensors data for rainfall and stream heights( https://www.data.brisbane.qld.gov.au/data/dataset/telemetry-sensors-rainfall-and-stream-heights). I thought this would be a great starting point for my IoT platform idea.

Building a platform from scratch is hard enough much more using a modern tech stack that I have not used before. So, my plan is to make it as simple as possible to make it into a Proof-of-Concept (PoC) where I can then make it robust in the future. So here are the technologies that I chose to use for this project. What is so great about these tools is that I can write all my code in Python which is very common nowadays in the data world.

• Prefect — For workflow orchestration
• DuckDB — An in-memory/in-process OLAP SQL database
• FastAPI — A very fast Python framework for creating APIs
• Streamlit — A Python web framework for creating Data Apps
• Docker — For containerization
• Azure Cloud — For deploying the platform

At the end of the project, this is what we can see from the dashboard

With so much technology to cover, I decided to break this article into multiple parts to give a more detailed explanation of each component. For this first part, we are going to discuss Prefect and DuckDB for data extraction and exploration.

A. Data Extract using Prefect

Upon quick investigation of the dataset site, I found out that there are two CSV files, one contains the metadata of the sensors themselves ( id, type, geolocation, measurement unit, etc), and the other the actual sensor readings for every 15 minutes collected from the device — this file is in wide format CSV where the columns are the sensor_ids and a timestamp and contains the readings for the past 24-hours with 10-mins windows. The figure below might explain better what the file I get looks like.

From the looks of it, I know I will have multiple duplicate records if I just blindly process and append the records to my raw table. So to minimize this duplication I need to clean them first before inserting them into the DuckDB table. But first I have to extract them from the dataset portal. To do that I just have to grab the download URL from the website and then use a Python script to download it.

Now to extract using Prefect we have to create ‘flows’ and ‘tasks’ functions, think of it as ‘DAGS’ for Airflow or Dagster. I like these terminologies because I can immediately relate to the meaning. I should probably have another in-depth article about Prefect one day to discuss all its features. The ‘flow’ decorator in the code below tells the Prefect agent the entry point to the script. Each flow can have multiple tasks and sub-flows but tasks can’t have subtasks inside it( make sense to me, ‘Single Responsibility’)

import db as db
import httpx
from datetime import datetime, date

from prefect import flow, task, get_run_logger,serve
from pathlib import Path
from dotenv import load_dotenv

DOWNLOAD_URL = "https://www.data.brisbane.qld.gov.au/data/dataset/01af4647-dd69-4061-9c68-64fa43bfaac7/resource/78c37b45-ecb5-4a99-86b2-f7a514f0f447/download/"

"""
Extract latest file from Brisbane Open Data Portal
https://www.data.brisbane.qld.gov.au/data/dataset/telemetry-sensors-rainfall-and-stream-heights
"""
@task
def get_gauge_data() -> str:

    now = datetime.today()
    date_str = str(now.date()) 
    date_str = date_str.replace("-","")
    print(date_str)

    filename = f"gauge-data-{date_str}*.csv"
    print("filename:" + filename)
    try:
        r = httpx.get(f'{DOWNLOAD_URL}{filename}')
    
        now_str= str(now).replace(" ","-")
        outfilename = f'src/data/staging/new-file-{now_str}.csv'

        with open(outfilename, 'w') as f:
            f.write(r.text)
    
        return outfilename
    
    except Exception as ex:
        print(ex)
        raise Exception(f"Something happen: {ex}")


@flow
def extract_file_flow(name="Gauge Data Extract", log_prints=True):
    logger = get_run_logger()
    now = datetime.now()
    print(now)
    logger.info(f"Run date🤓: - {now} -")

    filename = get_gauge_data()
    logger.info(f"New file extracted: {filename}")
    #merge_gauge_data(filename)

if __name__ == "__main__":

    extract_file_flow()

To deploy this to your local environment, you have to start first your local Prefect server. You will need to install Prefect first using ‘pip’ or any Python package manager you have.

# pip install prefect

# prefect server start

Now we can deploy our extract flow like this with parameters

• -n — your deployment name
• -q — work pool queue name
• interval — frequency schedule in seconds

prefect deployment build src/flows/extract_flow.py:extract_file_flow -n extract-file-dev-deploy-15mins -q default --interval 900

Yes! Our flow is now scheduled to be executed every 15 minutes. Prefect will handle all the flow scheduling and execution for us now. If you want to monitor this from the UI you can now head to http://127.0.0.1:4200/

B. Data Seeding to DuckDB

To be able to analyse the data we extracted from the Council website we needed a place where we can store them — this is where DuckDB comes in.

I decided beforehand that I would use a persistent DuckDB instead of the in-session memory so I could save the records in a relation state rather than reading multiples again. In this way, I can append new raw data that comes in and record in an audit table the last insert timestamp that the task processed.

To create the tables in my DuckDB instance, I downloaded multiple sample files from the source and then wrote a one-off Prefect flow to seed the database taking advantage of DuckDB’s auto-creation of tables from reading a CSV file. The sample code below lets me populate my database.

import db as db
import pandas as pd
import os

from duckdb import DuckDBPyConnection
from prefect import flow, task, get_run_logger,serve
from datetime import datetime, date
from pathlib import Path

####### HELPER FUNCTIONS ##########
def get_conn():
    return db.get_db_conn('db/datahack.duckdb')

def init_seq(seq_name):
   conn = get_conn()
   return conn.execute(f"CREATE SEQUENCE IF NOT EXISTS {seq_name} START 1;")

def init_sensor_table():
    conn = get_conn()
    #create metadata look up table
    conn.execute(
        """
            DROP TABLE IF EXISTS gauge_sensors_metadata; 
            CREATE TABLE gauge_sensors_metadata AS 
            SELECT * 
            FROM read_csv_auto
            ('src/data/metadata/rainfall-and-stream-heights-metadata-20230917t110000.csv')
        """
    )
    
def init_audit_table():
    conn = get_conn()
    conn.execute(
        """
            DROP TABLE IF EXISTS ingest_pipeline_audit_log ;
            CREATE TABLE ingest_pipeline_audit_log
            (
                auditid integer primary key DEFAULT NEXTVAL('seq_auditid'),
                ingest_filename varchar(255) not null,
                last_datetime datetime not null,
                updated_by varchar(255),
                updated_date datetime
            )
        """
    )
   

def insert_to_audit_table(record):
     conn = get_conn()
     query = f"""
        INSERT INTO ingest_pipeline_audit_log (
            ingest_filename, 
            last_datetime, 
            updated_by, 
            updated_date
        ) 
        VALUES {record}
     """
     return conn.execute(query)
 
def get_last_datetime():
    conn = get_conn()
    last_datetime = conn.sql(
        """
        SELECT last_datetime 
        FROM ingest_pipeline_audit_log
        WHERE auditid = 
            ( SELECT max(auditid) 
            FROM ingest_pipeline_audit_log
            )
        """
        ).fetchone()[0]
    
    print(f'last_datetime:{last_datetime}')
    return last_datetime


####### TASKS FUNCTIONS ##########

@task
def init_tables():
    
    init_seq("seq_auditid")
    init_sensor_table()
    init_audit_table()


@task
def display():
    conn = get_conn()
    conn.sql('SELECT * FROM ingest_pipeline_audit_log').show()
    conn.sql('SELECT * FROM ingest_gauge_metrics').show()

@task
def process_seeds(filepath: str):
    conn = get_conn()
    files =[ f for f in os.listdir(filepath) ]
    files_sorted = sorted(files)

    #Loop all through seed files
    for index, filename in enumerate(files_sorted):
        print(f"Processing --- {filename}" )
        filepath = Path("src/data/seeds") / filename
        tmp_df = conn.sql(f"SELECT * FROM read_csv_auto('{filepath}',all_varchar=true)").df()
        tmp_df["measured_datetime"] = pd.to_datetime(tmp_df['Measured'])
        tmp_df["measured_day"] = pd.to_datetime(tmp_df['Measured']).dt.date
        tmp_df["measured_hour"] = pd.to_datetime(tmp_df['Measured']).dt.hour
        tmp_df["measured_minute"] = pd.to_datetime(tmp_df['Measured']).dt.minute
        tmp_df = tmp_df.drop(["Measured"], axis=1)
        
        tmp_df = tmp_df.replace(['-'], '0')
        
        #Create the table on first file
        if index == 0:
            
            drop_table('ingest_gauge_metrics')
            conn.execute("CREATE TABLE ingest_gauge_metrics AS SELECT * FROM tmp_df")

            last_datetime = conn.execute('SELECT MAX(measured_datetime) FROM ingest_gauge_metrics').fetchone()[0]

            print('Initial ingest')
            print(f'Processing {filename}')
            print(f'last_datetime : {last_datetime}') 
           
            audit_entry = (filename, last_datetime.strftime('%Y-%m-%dT%H:%M:%S'), 'datamatiks',datetime.now().strftime('%Y-%m-%dT%H:%M:%S'))
            insert_to_audit_table(audit_entry)
        else:
            sensors_tbl_stg = conn.sql("SELECT * FROM tmp_df")
            curr_last_datetime = get_last_datetime()
            new_df = conn.sql(f"SELECT * FROM sensors_tbl_stg where measured_datetime > '{curr_last_datetime}'").df()

            conn.execute('INSERT INTO ingest_gauge_metrics SELECT * FROM new_df')
            new_last_datetime = conn.sql('SELECT MAX(measured_datetime) FROM sensors_tbl_stg').fetchone()[0]

            audit_entry = (filename, new_last_datetime.strftime('%Y-%m-%dT%H:%M:%S'),'datamatiks',datetime.now().strftime('%Y-%m-%dT%H:%M:%S'))
            insert_to_audit_table(audit_entry)

####### FLOW FUNCTIONS ##########
@flow
def init_db_flow(name="Initialize DB flow ", log_prints=True):
    logger = get_run_logger()
    now = datetime.now()
    print(now)
    logger.info("%s Run date🤓:", now)

    init_tables()
    process_seeds('src/data/seeds')
    display()

if __name__ == "__main__":
    init_db_flow()

In order to avoid duplicate records, I checked for the last_datetime I recorded in an audit table and then just read the records from the current data frame that has the last_datetime after the last recorded last_datetime. To ensure it is processing the correct order of files. I have sorted the list of filenames before parsing them.

With that, I was able to create my database now.

C. Exploring my data using DuckDB CLI

To explore my database, I installed the no-dependencies duckdb cli binary.

I intentionally put the binary file inside my project folder for convenient access to my duckdb file.

To use this cli, it is similar to any SQL console application out there. You use the familiar SQL commands to display and query tables.

In order to open a persistent database you should this command below :

D .open <dbfilename>.duckdb

More commands can be found in their documentation — https://duckdb.org/docs/archive/0.4.0/api/cli

That’s it for this part and please check out Part 2 of this article to learn more about FastAPI and Streamlit.

Thanks for reading.

UPDATE: GitHub repo: https://github.com/datamatiks/iot-platform-modern-data-stack-poc

Building an IoT Platform Using Modern Data Stack — Part 1 was originally published in Datamatiks on Medium, where people are continuing the conversation by highlighting and responding to this story.

Yes, you heard it right I actually did take a bus to work that time. Last year, about the same time as now, I moved my family to the sunny city of Brisbane, Queensland Australia from Sydney. I had an offer from a big travel company whose HQ is based here. It was still pre COVID-19 period then so it was okay to take public transport. My team here is using Azure Cloud Services to do Data Engineering however by that time I have not much experience with Azure so I have to learn and catch up fast. In order to do that, my manager told all of us to take Azure certifications.

This article lays out my strategy on how to pass the AZ-900 certification and that includes the daily bus ride to work I had(will come to this, promise ;D ). Of course, everyone has a different preparation style for a certain exam or certification. The strategy I will mention here works for me as I have been
applying them since college and it’s what works best for me.

Pre-review/study practice test

Ever since college, I have always applied this strategy when I am preparing for an exam/certification test. I will always first do a pre-review/study practice test before I embark on learning or studying about something. I will find some practice exam or practice test about the topic or certification I am going to take and then just go through it and try to answer it to the best of my knowledge, it does not matter if I have prior knowledge or not, I just want to evaluate what I know and I don’t know about something. The result tells me how much more I need to learn or study about it. It will also help me gauge how much improvement I have achieved during succeeding practice tests.

In the case of AZ-900, I researched three possible sites where I can have my practice test.

1. Official Practice Test by MindHub

If you go to the site of Microsoft certification, you will actually see a small link there that points you to its official practice test. However, when you click it, you will see that you actually have to pay at least $130 for 30 days of access. For me, this is a lot of money to spend so I opted not to try it. They do have the free demo but will only give you five practice questions. I have not tried their questions but their description says their questions are similar to the actual exam questions. So, if you are up to it, you can try them.

2. ITEXAMS.com

I learned about this site from a colleague who uses it on his practice test and review. I love that it has a free tier and has plenty of practice questions for AZ-900(192 questions to be exact). I must say their questions are closely similar to the actual exam questions. It also has an explanation for the correct answer. They seem to update their questions regularly as well.

What’s annoying though is it can only display 5 questions per page and you have to click a captcha to go to the next page. If you want to activate or unlock a custom appearance like more questions per page then you have to upgrade to a premium account. I can’t complain enough though, you get real exam questions using the free tier and that’s what matters.

They also don’t have the timer and mark for review features just like the actual exam. But who knows, maybe they will add it in the future.

3. Whizlabs

Finally, I stumbled upon Whizlabs while reading Quora and I never regretted using them. They offer cloud certification practice tests as well as other courses that will help you pass any certifications. What’s best is, it is cheaper than the previous two considering you don’t need a subscription to have access to their practice test.

They mock the actual exam scenario where you can mark questions for review and the timer feature helps you rate yourself if you are able to finish the actual exam on time or not.
So I decided to try them and used them for my pre-review/study practice test. It is really helpful because they tracked your score and the number of attempts to the practice test. That way you can see your progress.

Check them out if you are interested. They always give vouchers and freebies. And also, you have lifetime access to those practice tests, sweet! right?!

Downtime study/review session

Alright, so after my pre-review/study evaluation test. I now know where to focus more on my study or in this case review for AZ-900 certification. What is good about Whizlab is that it gives you a report of your practice test showing where you have more incorrect answers. Based on that I can focus my lesson/review plans.

Remember that bus ride that I mentioned above? This is where it comes to play. The bus ride to/from work is downtime for me. I have always tried to use my downtime for something productive or learn new things, so this where I usually insert my review and study for certifications.
I have 30 mins bus ride going to work and another 30 mins going home so I make use of them to study/review for the certifications. I allowed five days to study the AZ-900 Certification course in ACloudGuru by Lars Klint. I used to have access to the company subscription so I leverage on that.

So I will be seated at the back of the bus while watching on my phone Lars Klint explain the topics for Azure fundamentals. I skimmed those topics I already know about or make the video speed at 4X to those I am familiar with. I focus more on those topics I have low scores on my practice tests. For 1 week I did this during my bus ride as well as while waiting for the bus.

For the next week, it’s all about the practice test. I will do one practice in the morning bus ride to work and another practice test on the evening bus ride home. I accessed the Whizlab practice test using my phone, it has the same experience while doing it on the laptop. Sometimes I am so engaged in a practice test that I will miss my stop or took the wrong bus home. Taking the wrong bus home gave me more time to practice tests though so it’s not bad at all. I did this practice test until the scheduled exam is one day away.

Schedule when ready

What I meant by this is that whenever I felt that I am ready to take the certification then that is the time I will book or schedule the actual certification exams. When do I know I am ready to take it is based on my practice test results. When I see that I got consistently high marks on the practice for the last three or four practice tests then I believe I am ready.
I will schedule the exam and then just continue with my bus ride practice test until the exam date.

For AZ-900, you must at least get an 80% score on the exam to pass it. So to make sure I will pass it, I always aim to get at least 90% or above score during my practice test consistently. That way I have a bit of margin of confidence during the actual exam. I always aim high so if I missed my 90% target at least I will still pass the 80% exam passing score. It’s always good to set yourself a goal so that you know where you going and what needed to be done to achieve it.

Point of no return — Exam date!

During the exam date, I did one practice test in the morning and then relax the whole afternoon. By the way, I’ve chosen the home or office proctored test as this was the time all the testing centers were closed due to COVID-19. So, this is also my first time taking an exam at home. It is really weird because I have to choose a time in the middle of the night where everything is quiet and the internet is fast. The proctored ask me to show the whole room taking my laptop with me circling around my test area, to see if nobody is around. Once, all the checks were done, the exam started.

I made use of my time to really read and review questions before marking the answers. It also helps that my practice test questions were really similar to the real questions. It makes the test smooth and the experience similar to the practice test as well. At the end of the exam, you will get your result if you pass or not. Luckily I passed!

So, that’s it! hopefully, you learn something from my experience.

How I passed the AZ-900 Microsoft Azure Fundamentals test by riding the bus to work. was originally published in Datamatiks on Medium, where people are continuing the conversation by highlighting and responding to this story.

However, if you have been working on cloud platforms like AWS, GCP or Azure, it feels more natural to use tools and utilities that are native with your cloud provider.

Today, I want to introduce how you can connect to your Postgres DB instance in Azure as well as in your local docker instance using Azure Data Studio.

What is Azure Data Studio?

- Azure Data Studio is a cross-platform database tool for data professionals using on-premises and cloud data platforms on Windows, macOS, and Linux.

- Azure Data Studio offers a modern editor experience with IntelliSense, code snippets, source control integration, and an integrated terminal. It’s engineered with the data platform user in mind, with built-in charting of query result sets and customizable dashboards.

- The source code for Azure Data Studio and its data providers is available on GitHub under a source code EULA that provides rights to modify and use the software, but not to redistribute it or host it in a cloud service.

For more of it’s feature you can read it here Azure Data Studio

Download and Install Azure Data Studio

• Head over to download here to get the latest version of ADS that is suited to your development setup.
• Once you have installed it, just launched the app you should be able to see homepage similar as below.

Setting up Postgres docker instance

If you have not installed docker desktop before there is a good documentation on their site showing how to install docker on mac, windows and Linux . I will not be including it in this article but this guide will get you started.

For this article, we are going to use docker-compose which is included when you install docker desktop.

Below is a bare minimum docker-compose file that creates a postgres db instance in your local setup.

1. Create a directory for your postgres docker ( ex. mkdir postgres_docker)
2. Chage to that directory ( cd postgres_docker )
3. Create a file called docker-compose.yml. Paste the following code

version: '3'
services: 
  postgres:
    image: postgres
    restart: on-failure
    environment:
       - POSTGRES_USER=postgres
       - POSTGRES_PASSWORD=password123
       - PGDATA=/var/lib/postgresql/data/pgdata
       - POSTGRES_DB=postgres
    volumes:
       - ./postgres_data:/var/lib/postgresql
    ports:
       - 5432:5432

Here, we grab the postgres docker image, then set default database name(“postgres”) and default credentials ( username + password ).

We also set a volume in the local machine to make the data persistent even if we shutdown the docker instance. Then finally, mapping the localhost port to the port in the docker image.

Start Postgres docker image

Start docker image by running “docker-compose up” inside the directory where you save your docker-compose.yml file.

Connecting to Postgres from Azure Data Studio

pre-requisite

In order to access Postgres SQL from Azure Data Studio we first need to add the PostgresSQL extension which is still in Preview. After you install this then you can start connecting to Postgres already.

1. Now, go to your Azure Data Studio and add new connection.

2. Type the server credentials you have set in the docker-compose file.

3. Connected!

Testing and Querying

Now you can start playing around the GUI and write some queries

Bonus — Reporting and Extract

Some download options , so you can download the result the way you want it.

Simple Vizualization

You can also display the results with some default charting options.

That’s it! hope you learn something from this article and start using Azure Data Studio on your projects for accessing Postgres databases.

Using Azure Data Studio to connect to Postgres instance in Docker was originally published in Datamatiks on Medium, where people are continuing the conversation by highlighting and responding to this story.

For all the IoT enthusiasts out there I am sure you already heard the latest release of Raspberry Pi last year which the Raspberry Pi 4 Model B. If you have not heard about it nor have no idea about what is it then probably it’s good checking out this article — what is the raspberry pi 4? as this is not the focus of this article.

I received my Pi 4 as a Christmas gift from my sister in law back in the US but I have not it until this COVID-19 home isolation situation going on. At first, I only installed a Minecraft server so that I and my kids can play the hacked version of Minecraft while staying at home. But the kids got bored and leave the server lying around without no one using it.

So I decided I should try to run something else into it. It’s a good thing I am in the middle of learning about KSQL- Confluent’s streaming SQL engine that enables real-time data processing against Apache Kafka. This is good if you have some real-time analytics or IoT projects you want to build.

Below is a step-by-step guide on installing and running of Kafka and KSQL into Raspberry Pi 4 based on my own experience with notes and links to some issues I encountered during the process.

1. Setup your Raspberry Pi

I am not going to expand more about this but head to raspberrypi.org to know about preparing your Pi and installing OS. I would recommend using NOOBS and Raspbian OS for a starter.

2. Confluent Kafka/KSQL Installation

Once your Pi is ready, make sure you can connect to it using ssh so that you can easily access it from your laptop terminal.

To download the self-manage Confluent software go to https://www.confluent.io/download and download the appropriate software. You may need to register to get the free software.

Once you got the tar file downloaded, you can now secure transfer it to your Pi.

scp confluent-5.4.1–2.12.tar.gz pi@xxx.xxx.x.xx:/home/pi/

Next, access your Pi through ssh and uncompress the confluent tar file.

tar -xzvf confluent-5.4.1–2.12.tar.gz

There you have it, you have now the confluent software in your Pi. But wait! How do we run it? Good question.

3. Running Apache Kafka

By right, running Apache Kafka on a real server ( Linux machine/ Window Server) is simple. Confluent has a good utility to start Kafka and all its components automatically for us.

If you go to the recently created confluent folder and run this command:

bin/confluent start local

It will have an error instead of running all the services as shown below

Starting zookeeper
zookeeper is [UP]
Starting kafka
kafka is [UP]
Starting schema-registry
schema-registry is [UP]
Starting kafka-rest
kafka-rest is [UP]
Starting connect
connect is [UP]
Starting ksql-server
ksql-server is [UP]
Starting control-center
control-center is [UP]

The error looks like this below. It seems like ARM is not supported by this utility.

So, what I did is just execute it manually. I would recommend installing screen utility so that you can run the components in the background on its own screen.

sudo apt-get install screen

Now you can do this.

To run Apache Kafka, we first need to run the Zookeeper component. Think of it as a Cluster Manager, it manages and syncs all servers in a Kafka cluster. To manually start it, open a new screen with a name.

screen -S zookeeper

Once inside the screen, you can now start zookeeper. But first, change zookeeper dataDir to a persistent location not ‘/tmp’.

ex. dataDir=/var/zookeeper

bin/zookeeper-server-start etc/kafka/zookeeper.properties

To detach the screen and go back to the main screen just do Ctrl A+D ( for Mac Users ). To resume screen do :

screen -r zookeeper or screen -r <screen-id>

Now, do the same thing as above to run Kafka server. Modify the server property log.dirs to a persistent location too.

ex. ‘log.dirs=/var/kafka-logs’

screen -S kafka-server
bin/kafka-server-start etc/kafka/server.properties

If all goes well you now have a running Kafka Cluster with 1 broker and 1 zookeeper. Now we are ready to run KSQL server.

4. Testing your Kafka Cluster

Creating a topic

bin/kafka-topics --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic ksql-pi-demo-topic

Publishing to a topic

The simplest message we can publish to a topic is a string. But the most common use case in the real-world is a CSV file or delimited records. So let’s try that here. Just input a delimited record per line as seen below.

bin/kafka-console-producer --broker-list localhost:9092 --topic ksql-pi-demo-topic

>abc123,test123,nsw
>abc456,test456,qld
>abc789,test789,vic
>abc111,test111,qld

Consuming messages from a topic

Now we consume the record. We should see the exact sequence of data we published.

bin/kafka-console-consumer --bootstrap-server localhost:9092 --from-beginning --topic ksql-pi-demo-topic

abc123,test123,nsw
abc456,test456,qld
abc789,test789,vic
abc111,test111,qld

5. Running KSQL Server and CLI

KSQL server needs to access the schema registry which contains the schema of all the records for all topics that the Kafka cluster publishes. KSQL Server is like a group JVM nodes connecting to the Kafka cluster to process streams in a realtime manner using a familiar SQL like syntax. It is the query engine for streaming data in Kafka. To access those schemas we need to start the schema-registry.

bin/schema-registry-start etc/schema-registry/schema-registry.properties

After the schema-registry is up, we can start KSQL server. Since we are running from a Pi, we need to set some environment variables first so that we will not have and issue starting the server By default KSQL server needs a huge amount of RAM since it is doing its process in-memory. But for demo purpose in a Pi, we can set it at lower values.

export KSQL_HEAP_OPTS=”-Xms1G -Xmx2G”

I set it to that value so that I can still have more RAM left on my Pi. You change it according to your Pi’s spec. We can now start the KSQL Server.

KSQL_OPTS=”-Djava.io.tmpdir=/home/pi/javatmp/” bin/ksql-server-start etc/ksql/ksql-server.properties

Finally, when the server is up and running we can now start the KSQL-CLI to analyze streams of records from a topic.

bin/ksql http://0.0.0.0:8088

6. Creating and Testing KSQL Streams

To show the topics you have created in Kafka execute this command from the ksql-cli:

show topics;

Creating Stream from a topic

To be able to analyze streams in KSQL we have to create the stream first. It’s like a table that you can query.

ksql> CREATE STREAM ksqlpi_demo_streams (ID VARCHAR, NAME VARCHAR,STATE VARCHAR)
> WITH (kafka_topic=’ksql-pi-demo-topic’,value_format=’DELIMITED’);

Message
 — — — — — — — — 
 Stream created
 — — — — — — — —

Querying stream in realtime

This is how you can finally monitor all the records inside a Kafka topic using SQL.

select * from KSQLPI_DEMO_STREAMS EMIT CHANGES;

Below is a video showing KSQL in Action.

That’s all folk. Now you can do some amazing IoT stuff in your Raspberry Pi plus the power of Kafka and KSQL. I will continue to document my learning about KSQL and its features, so watch this space for more cool stuff.

Running Kafka and KSQL on your Raspberry Pi 4 was originally published in Datamatiks on Medium, where people are continuing the conversation by highlighting and responding to this story.

Life is an opportunity, benefit from it.
Life is beauty, admire it.
Life is bliss, taste it.
Life is a dream, realize it.
Life is a challenge, meet it.
Life is a duty, complete it.
Life is a game, play it.
Life is a promise, fulfill it.
Life is sorrow, overcome it.
Life is a song, sing it.
Life is a struggle, accept it.
Life is a tragedy, confront it.
Life is an adventure, dare it.
Life is luck, make it.
Life is too precious, do not destroy it.
Life is life, fight for it.

Mother Teresa

Originally published at nifrali.blogspot.com on November 29, 2018.

She had always wanted to breastfeed our children since our first born, but because of commitment to work and lack of milk produced, she had to stop when she had just already started and switched to formula. This time, she wanted to breastfeed the new baby again, but knowing her commitment to work she might face the same challenge of stopping her breastfeeding journey again. Although this time she is more prepared — she browses through the Internet to research ahead and see how other working mothers do it. It turns out that it is not only her having this problem — there a lot of mothers out there who wants to breastfeed but faces the same dilemma. Searching for solutions on the Internet like a marketplace or any local website to do logistics and deliver breast milk within one’s zone was a failure as there was none. She can only find Facebook groups or forums doing it. In Instagram, it’s even more difficult to find for non-tech-savvy moms. That’s why she thought of a possible solution to address this problem, which is to build a marketplace app for this market — an app that can do selling/buying/donating of breast milk in the Philippines, and maybe in the future the whole of Asia.

One day, I received an email from her, which stated, “I already have a solution! Build me this app, ASAP!” and I was like “Ahhh, okay, so when do you need it?” She replied, “next month!” Luckily, I have some mobile app development background so I scrambled some tech here and there to come out with a prototype and she liked it. Although I initially implemented this, I’m not going to do the entire development as she is going to hire her team of developers if this idea takes off because I also have my full-time job in an uber cool startup here in Singapore.

We all know that breastfeeding is best for babies as it boosts an infant’s immune system as well as prevents malnutrition. And as it turns out there is a big market in this niche area as we tried to dig more into it. There was a survey conducted by UNICEF in the Philippines, which shows that only 34% of infants below 6 months drink exclusive breast milk. The two most notable reasons for such are 1) mothers don’t have enough milk and 2) mothers are returning back to work. There really is an opportunity here for the mobile app to tap. Also, in terms of customer retention, the survey shows that 34% of those who breastfeed continues until the child is 2 years old. Of course, in terms of supply, a data from WHO will validate that this is a sustainable business model. Based on WHO website, “Of the 135 million babies born every year, only 42% are breastfed within the first hour of life, only 38% of mothers practice exclusive breastfeeding during the first six months and 58% of mothers continue breastfeeding up to the age of two years.”

We conducted a short poll in Cebu, Philippines to validate this. It’s a revelation that the top two reasons why mothers stop breastfeeding are the same reasons stated above. Furthermore, asking them if they are willing to sell or buy milk through a mobile app shows a huge 80% of the respondents saying yes. Also, 90% of them are in the age range of 25–34, which is mostly the Millennial and Gen Y who are educated and well informed.

The data above shows that there is a potential for this app to take off and can even scale — imagine deploying it to the whole Philippines or even in South Asia alone: Singapore, Indonesia, Malaysia.

This niche market is a goldmine ready to be tapped and HandyNanay is going to lead the way for it. The term “Nanay” is a Filipino equivalent for a mother.

Interested mothers can subscribe through the website www.handynanay.co and they will receive the first version of the app.