Energi dari Data Engineer

Menurut Head of Presales Dell Technologies Indonesia, Fitra Suryanto, hanya 12% dari organisasi yang mereka survey secara global mengerti dan mendapatkan insight dari data. Dewasa ini sering kali beberap orang mengatakan bahwa data is the new oil atau dengan kata lain data akan menjadi energi baru. Bahkan, beberapa orang menganggap data akan lebih mahal dibandingkan oil itu sendiri. Hal tersebut terjadi karena setiap organisasi pasti akan membuat keputusan dalam menjalankan roda bisnisnya, sedangkan keputusan itu perlu didukung oleh data yang akurat dan tepat.

Dalam dunia data ada istilah GIGO atau garbage in garbage out artinya jika data yang dimasukkan keliru maka keputusan yang dikeluarkan juga akan keliru. Oleh sebab itu untuk memastikan sebuah keputusan telah benar perlu dipastikan terlebih dahulu bahwa data yang diterima juga benar. Pada posisi ini orang yang bertugas menyediakan data menjadi vital. Secara sederhana tugas tersebut diemban oleh seorang data engineer. Ada 2 fundamental yang perlu dikuasai seorang data engineer, yaitu python dan SQL. Setelah fundamental keduanya dikuasai maka selanjutnya perlu untuk bisa menerapkannya pada data quality, data manipulation, dan data cleansing.

Ada 10 hal yang menjadi dasar dalam bahasa pemrograman Python, yaitu tipe data, aritmatika, comparison, logical, assignment, conditional, looping, data structure, function, dan class. Selebihnya merupakan penggunaan dari fitur yang telah ada di Python. Sedangkan di SQL ada 5 hal dasar yang perlu dikuasai, yaitu select, function, group by, inner join, dan union. Secara fundamental memang terlihat sederhana, yaitu hanya ada 15 hal yang perlu diketahui, namun jika dikombinasikan dalam hal yang sifatnya detail maka bisa ada ratusan kombinasi. Oleh sebab itu diperlukan kecerdikan untuk menggunakan tools yang ada di Python dan SQL. Output dari data engineer ada 3, yaitu:

Data Lake : data yang masih sesuai aslinya (belum ada manipulasi)
Data Warehouse : kumpulan seluruh data yang telah dimanipulasi
Data Marts : salah satu bagian dalam data warehouse untuk lini bisnis tertentu

Ketiga data tersebut akan dianalisa oleh data analyst kemudian machine learning engineer membuat mesin pembelajaran agar dapat mencegah fraud yang pernah terjadi misalnya. Nah, orang yang bisa melakukan tugas data engineer, data analyst, dan machine learning maka layak disebut sebagai data scientist. Jadi kerja seorang data engineer adalah mengumpulkan data dari berbagai sumber sedemikian sehingga data tersebut ditempatkan di data warehouse dan siap digunakan oleh data scientist. Dengan demikian ada 3 pertanyaan bagi seorang yang hendak menjadi data engineer, yaitu:

Bagaimana cara mendapatkan data, misalnya dari twitter?
Bagaimana cara memuat data sehingga dapat dibaca? (import)
Bagaimana cara memeriksa kualitas data yang telah didapatkan (apakah data tersebut memiliki impact terhadap bisnis)?
Bagaimana cara melakukan manipulasi dari data (agar lebih mudah dibaca)?
Bagaimana cara melakukan pembersihan data (baik missing value maupun outlier)?
Bagaimana cara menyimpan data yang telah berhasil diolah? (export)
Setelah data siap akan disimpan di mana? (on premise atau cloud)

Sumber: https://ilmudatapy.com/import-dan-export-data-di-python/

Cara Crawling

Untuk mengetahui apa yang sedang banyak diperbincangkan maka bisa digunakan twint untuk menarik kata kunci dari twitter. Gunakan python versi 3.8.5 dan install git bash. Buka git bash dan gunkan syntax di bawah:

$ pip install pipenv
$ pipenv --python 3.9
$ pipenv shell
$ pipenv install git+https://github.com/twintproject/twint.git#egg=twint

Setelah berhasil diinstall untuk melakukan crawling sebelumnya buka kembali git bash dan gunakan syntax:

$ pipenv shell
$ twint -u jokourno

Kemudian perhatikan git bash menuju ke folder mana, sebab folder tersebut yang akan dijadikan tempat penyimpanan dari data hasil crawling. Gunakan syntax di bawah untuk mendapatkan data:

$ twint -s "sambo" -o sambo.csv --limit 30 --since 2022-08-01 --until 2022-09-24 --csv

Keterangan dari command di atas adalah:

-s : mencari (search)
"sambo" : kata kunci yang dicari
-o : akan disimpan
sambo.csv : dengan nama file sambo
--limit 30 : cari sebanyak 30 data saja
--since 2022-08-01 : sejak tanggal 1 Agustus 2022
--until 2022-09-24 : hingga tanggal 24 September 2022
--csv : simpan dalam format csv

Cek di folder jika tidak ada masalah maka file sambo.csv akan muncul dan siap diakses baik melalui excel, calc, atau langsung melalui python. Selain crawling melalui twitter dengan twint, bisa juga dilakukan crawling di instagram dengan instalooter, serta crawling website.

Cara Import

Data yang telah didapatkan melalui crawling atau lainnya untuk diolah ke dalam python perlu dilakukan pembacaan atau import terlebih dahulu. Namun, dalam python istilah ini disebut load atau memuat. Sebelumnya install terlebih dahulu pandas untuk Python caranya adalah dengan menggunakan syntax:

pip install pandas

Lalu untuk membaca file csv gunakan syntax (hati-hati file harus berada di dalam folder yang digunakan untuk membuat kode di bawah):

import pandas as pd
dataset = pd.read_csv("sambo.csv")
print(dataset.head()) #untuk menampilkan 5 data teratas
print(dataset.head(10)) #untuk menampilkan 10 data teratas

Syntax di atas adalah contoh untuk melakukan load pada data dengan format .csv, untuk melakukan load pada data format lain bisa diperiksa di dokumentasi python. Cara import format lain:

Import .tsv

df_tsv = pd.read_csv("https://storage.googleapis.com/sample_tsv.tsv", sep='\t')
print(df_tsv.head(3)) # Menampilkan 3 data teratas

Import .xlsx

df_excel = pd.read_excel("https://storage.googleapis.com/sample_excel.xlsx", sheet_name="test")
print(df_excel.head(4)) # Menampilkan 4 data teratas

Import .json

url = "https://storage.googleapis.com/dqlab-dataset/covid2019-api-herokuapp-v2.json"
df_json = pd.read_json(url)
print(df_json.head(10)) # Menampilkan 10 data teratas

Import .sql

my_conn = mysql.connector.connect(host="relational.fit.cvut.cz",
port=3306,
user="guest",
passwd="relational",
database="financial",
use_pure=True)
my_query = """
SELECT *
FROM loan;
"""
df_loan = pd.read_sql_query(my_query, my_conn)
df_loan.head()
df_loan_= pd.read_sql(my_query, my_conn)
df_loan_.head()

Import google bigquery

query = """
SELECT *
FROM `bigquery-public-data.covid19_jhu_csse_eu.summary`
LIMIT 1000;
"""
df_covid19_eu_summary = pd.read_gbq(query, project_id="XXXXXXXX")
df_covid19_eu_summary

Cara Quality Check

Ada beberapa hal yang dilakukan untuk memeriksa kualitas dari data, yaitu tipe data, panjang kolom, banyak data yang tidak null, menghitung missing value, menghitung (min, max, mean, median, modus, standar deviasi), mencari quantile, menghitung korelasi, menangani missing value, menangani outlier, dan melakukan deduplikasi. Dengan demikian ada 10 prosedur standar yang digunakan untuk memeriksa kualitas dari data.

Tipe data

print(dataset.dtypes)

Panjang kolom

print(len(dataset['nama_kolom'])

Banyak data yang tidak null

print(dataset['nama_kolom'].count())

Menghitung missing value

missing_value = len(dataset['kolom1']) - dataset['kolom1'].count()
print(missing_value)

Menghitung (min, max, mean, median, modus, standar deviasi)

print(dataset['nama_kolom'].min())
print(dataset['nama_kolom'].max())
print(dataset['nama_kolom'].mean())
print(dataset['nama_kolom'].median())
print(dataset['nama_kolom'].mode())
print(dataset['nama_kolom'].std())

Mencari quantile

print(dataset['nama_kolom'].quantile([0.25, 0.5, 0.75]))

Menghitung korelasi

print(dataset[['kolom1', 'kolom2']].corr())

Menangani missing value

Memeriksa kolom yang memiliki missing value

print(dataset.isnull().any())

Imputasi kolom yang memiliki missing value (mengisi data kosong)

Mean

print(dataset['kolom_missing_value'].fillna(dataset.kolom_missing_value.mean()))

Modus

print(dataset['kolom_missing_value'].fillna(dataset.kolom_missing_value.modus()))

Predictive modeling

print(dataset['kolom_missing_value'].fillna(dataset.kolom_missing_value.function()))

Drop data yang memiliki missing value (membuang data kosong)

print(dataset['kolom_missing_value'].dropna())

Menangani outlier

Batas kontrol

Q1 = dataset['nama_kolom'].quantile(0.25)
Q3 = dataset['nama_kolom'].quantile(0.75)
IQR = Q3 - Q1

Hapus outlier

print('shape awal: ', dataset.shape)
dataset = [~((dataset['nama_kolom'] < (Q1 - 1.5 * IQR)) | (dataset['nama_kolom'] > (Q3 + 1.5 * IQR)))]
print('shape akhir: ', dataset.shape)

Melakukan deduplikasi

Periksa ukuran sebelum data duplikasi dibuang

print('Shape Awal: ', dataset.shape)

Buang data yang terduplikasi

dataset.drop_duplicates(inplace=True)

Periksa ukuran setelah data duplikasi dibuang

print('Shape Akhir: ', dataset.shape)

Dalam melakukan imputasi perlu dilakukan secara hati-hati, misalnya pada kolom id. Biasanya untuk mengangani missing value ini perlu dilakukan diskusi dengan business analyst atau dengan back end. Pada kasus kolom id ini tidak boleh dilakukan imputasi. Perlu dicatat juga untuk menangani missing value, ini bergantung pada adanya outlier atau tidak. Jika ada outlier atau pencilan ekstrim baik itu nilainya ekstrim tertinggi maupun ekstrim terendah maka tidak direkomendasikan menggunakan imputasi mean, sebab akan ada bias pada nilai mean-nya.

Cara Memanipulasi Data

Setelah data berhasil didapatkan melalui crawling, kemudian diimport ke python dan diperiksa kualitasnya maka selanjutnya akan dilakukan manipulasi data tersebut sedemikian rupa sehingga sesuai kebutuhan dari bisnis. Pandas adalah salah satu alat yang dapat digunakan untuk memanipulasi data. Di website pandas menyebut dirinya sebagai library dari python. Selain pandas biasanya untuk memperlancar proses, diinstal numpy yang juga merupakan library dari python. Ada 2 jenis kelas data yang perlu diketahui, yaitu data series dan data frame. Data series berupa array satu dimensi sedangkan dat frame berupa array dua dimensi (terdiri atas baris dan kolom). Maksud dimensi di sini adalah kolom dan baris.

Memeriksa data

print(dataset.info()) : hanya dapat digunakan untuk data frame
print(dataset.shape) : mengetahui berapa baris dan berapa kolom
print(dataset.dtypes) : mengetahui tipe data tiap kolom
print(dataset.astype("str")) : mengkonversi tipe data ke string
Membuat salinan:

dset = dataset.copy()
print(dset)

print(dataset.to_list()) : mengubah series menjadi struktur data list
print(dataset.unique()) : menghitung data yang unik (tidak duplikat)
print(dataset.index) : mencari index
print(dataset.columns) : mengetahui kolom yang ada di data frame
print(dataset.loc[0:1]) : melakukan slice berdasarkan nama kolom atau index
print(dataset.iloc[0:1]) : bisa untuk melakukan slice, mengubah, dan mengambil data

Mengkonversi data

Mengubah list menjadi series

ex_list = ['a',1,3,5,'c','d']
ex_series = pd.Series(ex_list)
print(ex_series)

Mengubah list of list menjadi data frame

ex_list_of_list = [[1,'a','b','c'],
[2.5,'d','e','f'],
[5,'g','h','i'],
[7.5,'j',10.5,'l']]
index = ['dq','lab','kar','lan']
cols = ['float','char','obj','char']
ex_df = pd.DataFrame(ex_list_of_list, index=index, columns=cols)
print(ex_df)

Mengubah dictionary menjadi series

dict_series = {'1':'a',
'2':'b',
'3':'c'}
ex_series = pd.Series(dict_series)
print(ex_series)

Mengubah dictionary menjadi data frame

df_series = {'1':['a','b','c'],
'2':['b','c','d'],
'4':[2,3,'z']}
ex_df = pd.DataFrame(df_series)
print(ex_df)

Mengubah numpy array menjadi series

arr_series = np.array([1,2,3,4,5,6,6,7])
ex_series = pd.Series(arr_series)
print(ex_series)

Mengubah numpy array menjadi data frame

arr_df = np.array([[1,2,3,5],
[5,6,7,8],
['a','b','c',10]])
ex_df = pd.DataFrame(arr_df)
print(ex_df)

Head & Tail

print("Tiga data teratas:\n", df.head(3))
print("Tiga data terbawah:\n", df.tail(3))

Indexing

df = pd.read_csv("https://storage.googleapis.com/sample_tsv.tsv", sep="\t")
print("Index:", df.index)
print("Columns:", df.columns)
Membuat index baru

df.index = ["Pesanan ke-" + str(i) for i in range(1, 11)]

print("Dataframe dengan index baru:\n", df)

Slicing

df = pd.read_csv("https://storage.googleapis.com/sample_csv.csv")
Slice berdasarkan kolom

df_slice = df.loc[(df["customer_id"] == 18055) &
(df["product_id"].isin(["P0029", "P0040", "P0041", "P0116", "P0117"]))]
print("Slice langsung berdasarkan kolom:\n", df_slice)

Slice berdasarkan index

df = df.set_index(["order_date", "order_id", "product_id"])
df_slice = df.loc[("2019-01-01",1612339,["P2154","P2159"]),:]
print("Slice df:\n", df_slice)

Transforming

df = pd.read_csv("https://storage.googleapis.com/sample_csv.csv")
Tampilkan tipe data:

print("Tipe data df:\n", df.dtypes)

Ubah tipe data pada kolom order

df["order_date"] = pd.to_datetime(df["order_date"])

Tampilkan tipe data setelah transformasi

print("\nTipe data df setelah transformasi:\n", df.dtypes)

Mengubah isi kolom menjadi lower case

df["brand"] = df["brand"].apply(lambda x: x.lower())

print("Kolom brand setelah apply:\n", df["brand"].head())
Mengambil karakter terakhir

df["brand"] = df["brand"].map(lambda x: x[-1])

print("Kolom brand setelah map:\n", df["brand"].head())

Inspeksi Missing Value

df = pd.read_csv("https://storage.googleapis.com/datacovid19.csv")
print(df.info())
Periksa missing value di setiap kolom

mv = df.isna().sum()
print("\nJumlah missing value per kolom:\n", mv)

print("Ukuran awal df: %d baris, %d kolom." % df.shape)
Mendrop (menghapus) missing value [alternatif 1]

Drop kolom yang seluruhnya missing value

df = df.dropna(axis=1, how="all")
print("Ukuran df setelah buang kolom dengan seluruh data missing: %d baris, %d kolom." % df.shape)

Drop baris jika ada satu saja data yang missing

df = df.dropna(axis=0, how="any")
print("Ukuran df setelah dibuang baris yang memiliki sekurangnya 1 missing value: %d baris, %d kolom." % df.shape)

Memfill (mengisi) missing value dengan string [alternatif 2]

print("Unique value awal:\n", df["province_state"].unique())
Isi missing value dengan "unknown_province_state"

df["province_state"] = df["province_state"].fillna("unknown_province_state")

print("Unique value setelah fillna:\n", df["province_state"].unique())

Memfill (mengisi) missing value dengan mean atau median [alternatif 3]

print("Awal: mean = %f, median = %f." % (df["active"].mean(), df["active"].median()))
print("Fillna median: mean = %f, median = %f." % (df_median.mean(), df_median.median()))
Isi missing value dengan nilai mean

df_mean = df["active"].fillna(df["active"].mean())

print("Fillna mean: mean = %f, median = %f." % (df_mean.mean(), df_mean.median()))

Memfill (mengisi) missing value dengan interpolasi [alternatif 4]

Data:

ts = pd.Series({
"2020-01-01":9,
"2020-01-02":np.nan,
"2020-01-05":np.nan,
"2020-01-07":24,
"2020-01-10":np.nan,
"2020-01-12":np.nan,
"2020-01-15":33,
"2020-01-17":np.nan,
"2020-01-16":40,
"2020-01-20":45,
"2020-01-22":52,
"2020-01-25":75,
"2020-01-28":np.nan,
"2020-01-30":np.nan
})

Isi missing value dengan interpolasi linier

ts = ts.interpolate()

print("Setelah diisi missing valuenya:\n", ts)

Tingkat lanjut

Append

Data series

s1 = pd.Series([1,2,3,4,5,6])
s2 = pd.Series(["a","b","c","d","e","f"])
s2_append_s1 = s2.append(s1)
print("Series - append:\n", s2_append_s1)

Data frame

df1 = pd.DataFrame({'a':[1,2],
'b':[3,4]})
df2 = pd.DataFrame({'b':[1,2],
'a':[3,4]})
df2_append_df1 = df2.append(df1)
print("Dataframe - append:\n", df2_append_df1)

Concat

df1 = pd.DataFrame({'a':[1,2],
'b':[3,4]})
df2 = pd.DataFrame({'b':[1,2],
'a':[3,4]})
Untuk baris

row_wise_concat = pd.concat([df2, df1])
print("Row-wise - concat:\n", row_wise_concat)

Untuk kolom

col_wise_concat = pd.concat([df2, df1], axis=1)
print("Column-wise - concat:\n", col_wise_concat)

Menambah identifier

multiindex_concat = pd.concat([df2, df1], axis=0, keys=['df2', 'df1'])
print("Multiindex - concat:\n", multiindex_concat)

Merge

df1 = pd.DataFrame({
'key':['k1','k2','k3','k4','k5'],
'val1':[200, 500, 0, 500, 100],
'val2':[30, 50, 100, 20, 10]
})
df2 = pd.DataFrame({
'key':['k1','k3','k5','k7','k10'],
'val3':[1,2,3,4,5],
'val4':[6,7,8,8,10]
})
Left join

merge_df_left = pd.merge(left=df2, right=df1, how='left', left_on='key', right_on='key')
print('Merge - Left:\n', merge_df_left)

Right join

merge_df_right = pd.merge(left=df2, right=df1, how='right', left_on='key', right_on='key')
print('Merge - Right:\n', merge_df_right)

Inner join

merge_df_inner = pd.merge(left=df2, right=df1, how='inner', left_on='key', right_on='key')
print('Merge - Inner:\n', merge_df_inner)

Outer join

merge_df_outer = pd.merge(left=df2, right=df1, how='outer', left_on='key', right_on='key')
print('Merge - Outer:\n', merge_df_outer)

Join

df1 = pd.DataFrame({
'key':['k1','k2','k3','k4','k5'],
'val1':[200, 500, 0, 500, 100],
'val2':[30, 50, 100, 20, 10]
})
df2 = pd.DataFrame({
'key':['k1','k3','k5','k7','k10'],
'val3':[1,2,3,4,5],
'val4':[6,7,8,8,10]
})
Menggunakan set_index dan keyword how

join_df = df1.set_index('key').join(df2.set_index('key'), how='outer')
print(join_df)

Pivot

data = pd.DataFrame({
'kelas': 6*['A'] + 6*['B'],
'murid': 2*['A1'] + 2*['A2'] + 2*['A3'] + 2*['B1'] + 2*['B2'] + 2*['B3'],
'pelajaran': 6*['math','english'],
'nilai': [90,60,70,85,50,60,100,40,95,80,60,45]
}, columns=['kelas','murid','pelajaran','nilai'])
Pivot dengan satu kolom

pivot1 = data.pivot(index='murid', columns='pelajaran', values='nilai')
print('Pivoting with single column measurement:\n', pivot1)

Pivot dengan banyak kolom

pivot2 = data.pivot(index='murid', columns='pelajaran')
print('Pivoting with multiple column measurement:\n', pivot2)

Pivot Tabel

data = pd.DataFrame({
'kelas': 6*['A'] + 6*['B'],
'murid': 2*['A1'] + 2*['A2'] + 2*['A3'] + 2*['B1'] + 2*['B2'] + 2*['B3'],
'pelajaran': 6*['math','english'],
'nilai': [90,60,70,85,50,60,100,40,95,80,60,45]
}, columns=['kelas', 'murid', 'pelajaran', 'nilai'])
Menggunakan mean

pivot_tab_mean = data.pivot_table(index='kelas', columns='pelajaran', values='nilai', aggfunc='mean')
print('Creating pivot table -- aggfunc mean:\n', pivot_tab_mean)

Menggunakan median

pivot_tab_median = data.pivot_table(index='kelas', columns='pelajaran', values='nilai', aggfunc='median')
print('Creating pivot table -- aggfunc median:\n', pivot_tab_median)

Melt

data = pd.DataFrame({
'kelas': 6*['A'] + 6*['B'],
'murid': 2*['A1'] + 2*['A2'] + 2*['A3'] + 2*['B1'] + 2*['B2'] + 2*['B3'],
'pelajaran': 6*['math','english'],
'nilai': [90,60,70,85,50,60,100,40,95,80,60,45]
}, columns=['kelas','murid','pelajaran','nilai'])
Membuat pivot

data_pivot = data.pivot_table(index='kelas', columns='pelajaran', values='nilai', aggfunc='mean').reset_index()
print('Pivoting dataframe:\n', data_pivot)

Melting pivot

data_melt_1 = pd.melt(data_pivot)
print('Melting dataframe:\n', data_melt_1)

Melting pivot dengan id_vars

data_melt_2 = pd.melt(data_pivot, id_vars='kelas')
print('Melting dataframe dengan idvars:\n', data_melt_2)

Unstack

data = pd.DataFrame({
'kelas': 6*['A'] + 6*['B'],
'murid': 2*['A1'] + 2*['A2'] + 2*['A3'] + 2*['B1'] + 2*['B2'] + 2*['B3'],
'pelajaran': 6*['math','english'],
'nilai': [90,60,70,85,50,60,100,40,95,80,60,45]
}, columns=['kelas','murid','pelajaran','nilai'])
Mengatur index

data = data.set_index(['kelas', 'murid', 'pelajaran'])
print('Dataframe multi index:\n', data)

Melakukan unstacking

data_unstack_1 = data.unstack()
print('Unstacking dataframe:\n', data_unstack_1)

Melakukan unstacking dengan level nama

data_unstack_2 = data.unstack(level='murid')
print('Unstacking dataframe dengan level name:\n', data_unstack_2)

Melakukan unstacking dengan level position

data_unstack_3 = data.unstack(level=1)
print('Unstacking dataframe dengan level position:\n', data_unstack_3)

Aggregation

gaq = pd.read_csv('https://storage.googleapis.com/global_air_quality_4000rows.csv')
Mean dan std

Membuat variabel pollutant

pollutant = gaq[['country', 'city', 'pollutant', 'value']].pivot_table(index=['country', 'city'],columns='pollutant').fillna(0)
print('Data pollutant (5 teratas):\n', pollutant.head())

Agregasi mean

pollutant_mean = pollutant.groupby('country').mean()
print('Rata-rata pollutant (5 teratas):\n', pollutant_mean.head())

Agregasi std

pollutant_std = pollutant.groupby('country').std().fillna(0)
print('Standar deviasi pollutant (5 teratas):\n', pollutant_std.head())

.sum() dan .unique()

Agregasi .sum()

pollutant_sum = pollutant.groupby('country').sum()
print('Total pollutant (5 teratas):\n', pollutant_sum.head())

Agregasi .unique()

pollutant_nunique = pollutant.groupby('country').nunique()
print('Jumlah unique value pollutant (5 teratas):\n', pollutant_nunique.head())

Multiple aggregation

Agregasi min, median, mean, max

multiagg = pollutant.groupby('country').agg(['min', 'median', 'mean', 'max'])
print('Multiple aggregations (5 teratas):\n', multiagg.head())

Custom aggregation

Buat fungsi IQR

def iqr(series):
Q1 = series.quantile(0.25)
Q3 = series.quantile(0.75)
return Q3-Q1

Agregasi IQR

custom_agg = pollutant.groupby('country').agg(iqr)
print('Custom aggregation (5 teratas):\n', custom_agg.head())

Custom aggregation by dict

custom_agg_dict = pollutant['value'][['pm10', 'pm25', 'so2']].groupby('country').agg({
'pm10':'median',
'pm25':iqr,
'so2':iqr
})
print('\nCetak 5 data teratas custom_agg_dict:\n', custom_agg_dict.head())

Time series

gaq = pd.read_csv('https://storage.googleapis.com/dqlab-dataset/LO4/global_air_quality_4000rows.csv')
gaq['timestamp'] = pd.to_datetime(gaq['timestamp'])
gaq = gaq.set_index('timestamp')
print('Dataset sebelum di-downsampling (5 teratas):\n', gaq.head())
Downsampling

Dari harian ke mingguan

gaq_weekly = gaq.resample('W').max()
print('Downsampling daily to weekly - max (5 teratas):\n', gaq_weekly.head())

Upsampling

Dari harian ke per jam

gaq_hourly = gaq.resample('H').mean()
print('Upsampling daily to hourly - mean (5 teratas):\n', gaq_hourly.head())

Visualisasi

Membuat pivot tabel yang menunjukkan waktu di baris

gaq_viz = gaq[['pollutant','value']].reset_index().set_index(['timestamp', 'pollutant'])
gaq_viz = gaq_viz.pivot_table(index='timestamp', columns='pollutant', aggfunc='mean').fillna(0)
gaq_viz.columns = gaq_viz.columns.droplevel(0)
print('Data (5 teratas):\n', gaq_viz.head())

Membuat fungsi

def default_val(val):
if val < 0:
return 0
else:
return val
line1 = gaq_viz.resample('M').mean().ffill().applymap(lambda x: default_val(x)).apply(lambda x: x/x.max()) # default value if value < 0 then 0, kemudian menghasilkan % value = value/max(value)
line1.plot(
title = 'average value of each pollutant over months',
figsize = (10,10), #ukuran canvas 10px x 10px
ylim = (0,1.25), #memberikan batas tampilan y-axis hanya 0 sampai 125%
subplots = True #memecah plot menjadi beberapa bagian sesuai dengan jumlah kolom
)
plt.ylabel('avg pollutant (%)')
plt.xlabel('month')
plt.show()

Pada bagian manipulasi data inilah peran seorang data engineer menjadi krusial. Diperlukan trik serta kecerdikan untuk menyiapkan data sesuai kebutuhan bisnis dengan bentuk data mentah yang masih teracak bahkan data yang masih bersifat unstructured.

Cara Cleansing Data

Melakukan data cleansing adalah bagian dari memperbaiki kualitas data. Pada proses pemeriksaan kualitas disinggung sedikit mengenai bagaimana menangani missing value dan dan outlier. Perbedaannya data cleansing dilakukan setelah data dilakukan manipulasi.

Mencari identitas yang valid

df_load = pd.read_csv('https://storage.googleapis.com/dataset.csv')
Jumlah baris dan kolom

print(df_load.shape)

Jumlah ID yang unik

print(df_load.customerID.nunique())

Memfilter ID format tertentu

df_load['valid_id']= df_load['customerID'].astype(str).str.match(r'(45\d{9,10})')
df_load = (df_load[df_load['valid_id'] == True]).drop('valid_id', axis=1)
print('Hasil jumlah ID Customer yang terfilter adalah', df_load['customerID'].count())

Memfilter ID duplikasi

Drop baris yang duplikat

df_load.drop_duplicates()

Drop duplikasi ID yang terurut berdasarkan periode

df_load = df_load.sort_values('UpdatedAt', ascending=False).drop_duplicates('customerID')
print('Hasil jumlah ID Customer yang sudah dihilangkan duplikasinya (distinct) adalah',df_load['customerID'].count())

Mengatasi missing value

Dengan menghapus

df_load['valid_id'] = df_load['customerID'].astype(str).str.match(r'(45\d{9,10})')
df_load = (df_load[df_load['valid_id'] == True]).drop('valid_id', axis = 1)
df_load.drop_duplicates()
df_load = df_load.sort_values('UpdatedAt', ascending=False).drop_duplicates('customerID')
print('Total missing values data dari kolom Churn', df_load['Churn'].isnull().sum())
Drop semua baris dengan kolom spesifik

df_load.dropna(subset=['Churn'], inplace=True)
print('Total Rows dan kolom Data setelah dihapus data Missing Values adalah', df_load.shape)

Dengan mengisi

df_load['valid_id'] = df_load['customerID'].astype(str).str.match(r'(45\d{9,10})')
df_load = (df_load[df_load['valid_id'] == True]).drop('valid_id', axis = 1)
df_load.drop_duplicates()
df_load = df_load.sort_values('UpdatedAt', ascending=False).drop_duplicates('customerID')
df_load.dropna(subset=['Churn'],inplace=True)
print('Status Missing Values :',df_load.isnull().values.any())
print('\nJumlah Missing Values masing-masing kolom, adalah:')
print(df_load.isnull().sum().sort_values(ascending=False))
Mengisi missing value dengan angka 11

df_load['tenure'].fillna(11, inplace=True)

Mengisi missing value dengan median

for col_name in list(['MonthlyCharges', 'TotalCharges']):
median = df_load[col_name].median()
df_load[col_name].fillna(median, inplace=True)
print('\nJumlah Missing Values setelah di imputer datanya, adalah:')
print(df_load.isnull().sum().sort_values(ascending=False))

Mengatasi outlier

Mendeteksi outlier

df_load['valid_id'] = df_load['customerID'].astype(str).str.match(r'(45\d{9,10})')
df_load = (df_load[df_load['valid_id'] == True]).drop('valid_id', axis = 1)
df_load.drop_duplicates()
df_load = df_load.sort_values('UpdatedAt', ascending=False).drop_duplicates('customerID')
df_load.dropna(subset=['Churn'],inplace=True)
df_load['tenure'].fillna(11, inplace=True)
for col_name in list(['MonthlyCharges','TotalCharges']):
median = df_load[col_name].median()
df_load[col_name].fillna(median, inplace=True)
print('\nPersebaran data sebelum ditangani Outlier: ')
print(df_load[['tenure','MonthlyCharges','TotalCharges']].describe())
Membuat box plot

plt.figure()
sns.boxplot(x=df_load['tenure'])
plt.show()
plt.figure()
sns.boxplot(x=df_load['MonthlyCharges'])
plt.show()
plt.figure()
sns.boxplot(x=df_load['TotalCharges'])
plt.show()

Mengatasi outlier

df_load['valid_id'] = df_load['customerID'].astype(str).str.match(r'(45\d{9,10})')
df_load = (df_load[df_load['valid_id'] == True]).drop('valid_id', axis = 1)
df_load.drop_duplicates()
df_load = df_load.sort_values('UpdatedAt', ascending=False).drop_duplicates('customerID')
df_load.dropna(subset=['Churn'],inplace=True)
df_load['tenure'].fillna(11, inplace=True)
for col_name in list(['MonthlyCharges','TotalCharges']):
median = df_load[col_name].median()
df_load[col_name].fillna(median, inplace=True)
Q1 = (df_load[['tenure','MonthlyCharges','TotalCharges']]).quantile(0.25)
Q3 = (df_load[['tenure','MonthlyCharges','TotalCharges']]).quantile(0.75)
IQR = Q3 - Q1
maximum = Q3 + (1.5*IQR)
print('Nilai Maximum dari masing-masing Variable adalah: ')
print(maximum)
minimum = Q1 - (1.5*IQR)
print('\nNilai Minimum dari masing-masing Variable adalah: ')
print(minimum)
more_than = (df_load > maximum)
lower_than = (df_load < minimum)
df_load = df_load.mask(more_than, maximum, axis=1)
df_load = df_load.mask(lower_than, minimum, axis=1)
print('\nPersebaran data setelah ditangani Outlier: ')
print(df_load[['tenure','MonthlyCharges','TotalCharges']].describe())

Menstandardisasi nilai

df_load['valid_id'] = df_load['customerID'].astype(str).str.match(r'(45\d{9,10})')
df_load = (df_load[df_load['valid_id'] == True]).drop('valid_id', axis = 1)
df_load.drop_duplicates()
df_load = df_load.sort_values('UpdatedAt', ascending=False).drop_duplicates('customerID')
df_load.dropna(subset=['Churn'],inplace=True)
df_load['tenure'].fillna(11, inplace=True)
for col_name in list(['MonthlyCharges','TotalCharges']):
median = df_load[col_name].median()
df_load[col_name].fillna(median, inplace=True)
Q1 = (df_load[['tenure','MonthlyCharges','TotalCharges']]).quantile(0.25)
Q3 = (df_load[['tenure','MonthlyCharges','TotalCharges']]).quantile(0.75)
IQR = Q3 - Q1
maximum = Q3 + (1.5*IQR)
minimum = Q1 - (1.5*IQR)
more_than = (df_load > maximum)
lower_than = (df_load < minimum)
df_load = df_load.mask(more_than, maximum, axis=1)
df_load = df_load.mask(lower_than, minimum, axis=1)
for col_name in list(['gender','Dependents','Churn']):
print('\nUnique Values Count \033[1m' + 'After Standardized \033[0mVariable', col_name)
print(df_load[col_name].value_counts())

Data cleansing berfokus pada 3 hal utama yaitu missing value, outlier, dan standarisasi nilai. Selama tidak ada outlier maka missing value masih memungkinkan untuk diisi dengan mean. Beberapa ahli data merekomendasikan nilai median jika terdapat outlier. Namun, dalam hal menangani missing value harus didiskusikan dengan tim terkait seperti business intelligent atau backend atau machine learning. Dapat dikatakan proses cleansing merupakan sentuhan terakhir terhadap data sebelum diekspor. Data cleaning itu ada 2: structured dan unstructured, khusus data yang unstructured maka perlu melewati cara seperti NER, POS, Lemmatization, etc.

Cara Export Data

Setelah melewati pemeriksaan kualitas data kemudian data dimanipulasi sesuai kebutuhan dan diakhiri dengan cleansing maka data akan diekspor atau di save dalam format yang disepekati.

df = ...
Menyimpan ke .csv

df.to_csv("csv1.csv", index=False)

Menyimpan ke .tsv

df.to_tsv("tsv1.tsv", index=False, sep='\t')

Menyimpan ke .clipboard

df.to_clipboard()

Menyimpan ke .xlsx

df_excel.to_excel("xlsx1.xlsx", index=False)

Menyimpan ke .gbq

df.to_gbq("temp.test", project_id="XXXXXX", if_exists="fail")

Pada dasarnya ekspor data ini dilakukan untuk menyimpan data dalam format tertentu. Keberhasilan dari big data juga diperhatikan melalui 5V yaitu volume, velocity, variety, veracity dan value (Storey & Song, 2017).

Cara Save Data

Setelah big data ditentukan formatnya maka hal yang krusial selanjutnya adalah dimana penyimpanan data tersebut. Ada 2 alternatif, yaitu on-premise dan cloud. Kalau penyimpanan dilakukan di on-premise apakah data disimpan begitu saja atau perlu dilakukan teknik kriptografi yang memberikan tingkat keamanan lebih baik. Sedangkan jika disimpan di cloud, biasanya layananan keamanan tertentu telah tersedia. Salah satu layanan data warehouse adalah redshift di Amazon atau yandex management service di Yandex.

Dengan demikian minimal ada 7 job description bagi seorang data engineer, yaitu crawling, import, quality, manipulation, cleansing, export, and save.

ref:

katadata.co.id

tantut-wahyuu-s

arief-luqman

Jokourno

Cari Blog Ini

Energi dari Data Engineer

Komentar