Scribd
Upload
8 views2 pages

Data Pipeline for JSON to CSV Conversion

The document contains a Python ETL pipeline consisting of four scripts: extract, transform, load, and main. The extract script retrieves user data from a JSON API, the transform script cleans the data by dropping missing usernames and renaming a column, and the load script saves the transformed data to a CSV file. The main script orchestrates the entire process, logging each step of the pipeline.

Uploaded by

akshay.app2011
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as PDF, TXT or read online on Scribd
8 views2 pages

Data Pipeline for JSON to CSV Conversion

The document contains a Python ETL pipeline consisting of four scripts: extract, transform, load, and main. The extract script retrieves user data from a JSON API, the transform script cleans the data by dropping missing usernames and renaming a column, and the load script saves the transformed data to a CSV file. The main script orchestrates the entire process, logging each step of the pipeline.

Uploaded by

akshay.app2011
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as PDF, TXT or read online on Scribd

EXTRACT.

PY
import pandas as pd
import requests
import logging
def extract_data():
try:
[Link](f"extracting file")
response =[Link]("[Link]
users")
data_df = [Link]()
df = [Link](data=data_df)
#print(df)
[Link]("JSON file successfully extracted")
return df
except Exception as e:
[Link](f" unable to read")
raise

[Link]
import pandas as pd

def transform_data(df:[Link]):

df = [Link](subset=['username'])

df = [Link](columns={'name':'Name'})

return df

[Link]
import pandas as pd
import logging

def load_data(df:[Link] , output_path):


df.to_csv(output_path,index=False , header=True)
[Link](f"file successfully loaded at {output_path}")
[Link]
import pandas as pd
import logging
import extract_API
import transform_API
import load_API

def setupLogger():
[Link](
filename= '[Link]',
format = '%(asctime)s - %(name)s - %(levelname)s - %(message)s',
level= [Link]
)
[Link]().addHandler([Link]())

def main():
setupLogger()

output_file_path = '/Users/akshay/Desktop/Data Engineering And AI /


Project 2/JSON_data_output.csv'

[Link]("Pipeline started")
extracted_df = extract_API.extract_data()
[Link]("extraction completed")
transformed_df = transform_API.transform_data(extracted_df)
[Link]('transformation completed')
load_API.load_data(transformed_df , output_file_path)

# if __name__ == '__main__':
main()

You might also like