A Comprehensive Guide to Importing Data into Singdata Lakehouse

Data Ingestion: Loading Files from the Web into the Lakehouse via Singdata Lakehouse Studio's Built-in Python Node

Overview

Singdata Lakehouse Studio comes with a built-in Python node that allows you to develop and run Python code.

Use Case

This is suitable for scenarios where third-party Python libraries need to be called to process files during data ingestion. For example, in this case, the Python library for Alibaba Cloud Object Storage Service (OSS) is called.

Implementation Steps

Create a New Python Task

Navigate to Development -> Tasks, click "+", and create a new Python task.

Task Name: 05_Loading Files from the Web into the Lakehouse via Studio's Built-in Python Node.

Develop Python Task Code

Paste the following code into the code editor of the newly created Python task:

import os,io import subprocess from datetime import datetime, timedelta import oss2

Alibaba Cloud OSS configuration:

ACCESS_KEY_ID = '${ak}' ACCESS_KEY_SECRET = '${sk}' BUCKET_NAME = 'yourbucketname' ENDPOINT = 'oss-cn-hangzhou-internal.aliyuncs.com' ROOT_PATH = f'{BUCKET_NAME}/ingest_demo/from_web' try: # Construct wget command url = f"https://github.com/yunqiqiliang/clickzetta_quickstart/blob/main/a_comprehensive_guide_to_ingesting_data_into_clickzetta/data/lift_tickets_data.csv.gz" cmd = ["wget", "-qO-", url] print(f"wget cmd: {cmd}") # Execute wget command and capture output wget_output = subprocess.check_output(cmd) print(f"Wget file done...") # Convert output to an in-memory file object file_obj = io.BytesIO(wget_output) except Exception as e: print(f"An error occurred: {e}") file_obj = None raise if file_obj: try: # Initialize Alibaba Cloud OSS auth = oss2.Auth(ACCESS_KEY_ID, ACCESS_KEY_SECRET) bucket = oss2.Bucket(auth, ENDPOINT, BUCKET_NAME) # Upload file to OSS oss_path = f"{ROOT_PATH}/lift_tickets_data.csv.gz" print(f"osspath: {oss_path}") bucket.put_object(oss_path, file_obj) print(f"Put file to oss done...") except Exception as e: print(f"An error occurred: {e}") finally: # Close the in-memory file object file_obj.close()

Configure Parameters for the Task

There are two parameters:

ACCESS_KEY_ID = '${ak}'ACCESS_KEY_SECRET = '${sk}'

Click on the schedule to fill in the default values for the parameters.

Click "Load Parameters from Code" and fill in the corresponding values.

Run the Test

Click "Run" to execute the Python code.

Check the Upload Results

Next Steps

Schedule Python tasks to achieve periodic data lake ingestion.
Analyze the files loaded into the data lake using SQL.
Form a complete ELT workflow with other tasks.

Resources

Studio Python Task Node