Yunqi Lakehouse

https://yunqi.tech/documents/
Admin
Yunqi Lakehouse is a next-generation cloud data lakehouse platform with incremental computing that...
Tokens:2,727
Snippets:7
Trust Score:1.2
Update:9 months ago
Show doc for...
Tokens:
Raw
### SQL JSON Extraction Examples

Source: https://yunqi.tech/documents/json_analyze

This SQL query demonstrates various ways to extract data from JSON objects stored in a table. It uses `json_extract` to access nested fields, array elements, and wildcards. The input JSONs vary in structure to show how the query handles missing or different fields.

```sql
SELECT json_extract(j, '$.*.city'),
       json_extract(j, '$.phoneNumbers[*]'),
       json_extract(j, '$.phoneNumbers[*].extra'),
       json_extract(j, '$.phoneNumbers[*].extra[*]'),
       json_extract(j, '$.*[*].extra[*]')
FROM (SELECT parse_json(s) as j
      FROM VALUES ('{ "firstName": "John", "lastName": "doe", "age": 26, "address": { "streetAddress": "naist street", "city": "Nara", "postalCode": "630-0192" }, "phoneNumbers": [ { "type": "iPhone", "number": "0123-4567-8888" }, { "type": "home", "number": "0123-4567-8910", "extra": [] } ] }'),
                  ('{ "firstName": "Ada", "lastName": "doe", "age": 20, "address": { "streetAddress": "naist street", "city": "Nara", "postalCode": "630-0192" }, "address2": {"city": "NewYork"}, "phoneNumbers": [ { "type": "iPhone", "number": "0123-4567-8888" }, { "type": "home", "number": "0123-4567-8910", "extra": [1,2,3] } ] }'),
                  ('{} ') as t(s));
```

--------------------------------

### Create Staging Tables for Data Pipeline

Source: https://yunqi.tech/documents/MigrateSnowflakeRealtimeETLPipelinetoClickzettaLakehouse

SQL statements to create staging tables for normalized data processed by the data pipeline. These tables are intended for intermediate storage before final transformation.

```sql
CREATE  TABLE IF NOT EXISTS staging_policy(
    policy_id VARCHAR(15) ,
    subscription_length FLOAT,
    region_code VARCHAR(5),
    segment VARCHAR(10));
    
CREATE  TABLE IF NOT EXISTS staging_vehicles (
    vehicle_id VARCHAR(15) ,
    vehicle_age FLOAT,
    fuel_type VARCHAR(10),
    parking_sensors VARCHAR(5),
    parking_camera VARCHAR(5),
    rear_brakes_type VARCHAR(10),
    displacement INT,
    trasmission_type VARCHAR(20),
    steering_type VARCHAR(15),
    turning_radius FLOAT,
    gross_weight INT,
    front_fog_lights VARCHAR(5),
    rear_window_wiper VARCHAR(5),
    rear_window_washer VARCHAR(5),
    rear_window_defogger VARCHAR(5),
    brake_assist VARCHAR(5),
    central_locking VARCHAR(5),
    power_steering VARCHAR(5),
    day_night_rear_view_mirror VARCHAR(5),
    is_speed_alert VARCHAR(5),
    ncap_rating INT);
  
CREATE  TABLE IF NOT EXISTS staging_customers(
    customer_id VARCHAR(20) ,
    customer_age INT,
    region_density INT);
  
CREATE  TABLE IF NOT EXISTS staging_claims(
    claim_id VARCHAR(20) ,
    claim_status INT);
```

--------------------------------

### Download Kaggle Data and Upload to S3 in Python

Source: https://yunqi.tech/documents/MigrateSnowflakeRealtimeETLPipelinetoClickzettaLakehouse

This Python script automates the process of downloading a dataset from Kaggle, extracting a specific CSV file, sampling the data using `load_random_sample`, and uploading the processed CSVs to an AWS S3 bucket. It requires Kaggle API credentials and AWS credentials to be configured.

```python
import pandas as pd
import boto3
import random
import os, json, io
import zipfile
from datetime import datetime
import subprocess
import sys
import contextlib

@contextlib.contextmanager
def suppress_stderr():
    with open(os.devnull, 'w') as devnull:
        old_stderr = sys.stderr
        sys.stderr = devnull
        try:
            yield
        finally:
            sys.stderr = old_stderr

def load_random_sample(csv_file, sample_size):

    # Count total rows in the CSV file
    total_rows = sum(1 for line in open(csv_file, encoding='utf-8')) - 1  # Subtract header row

    # Calculate indices of rows to skip (non-selected)
    skip_indices = random.sample(range(1, total_rows + 1), total_rows - sample_size)

    # Load DataFrame with random sample of rows
    df = pd.read_csv(csv_file, skiprows=skip_indices)

    policy_table = df[['policy_id', 'subscription_length', 'region_code', 'segment']].copy()
    vehicles_table = df[['policy_id', 'vehicle_age', 'fuel_type', 'is_parking_sensors', 'is_parking_camera', 'rear_brakes_type', 'displacement', 'transmission_type', 'steering_type', 'turning_radius', 'gross_weight', 'is_front_fog_lights', 'is_rear_window_wiper', 'is_rear_window_washer', 'is_rear_window_defogger', 'is_brake_assist', 'is_central_locking', 'is_power_steering', 'is_day_night_rear_view_mirror', 'is_speed_alert', 'ncap_rating']].copy()
    customers_table = df[['policy_id', 'customer_age', 'region_density']].copy()
    claims_table = df[['policy_id', 'claim_status']].copy()

    vehicles_table.rename(columns={'policy_id': 'vehicle_id'}, inplace=True)
    customers_table.rename(columns={'policy_id': 'customer_id'}, inplace=True)
    claims_table.rename(columns={'policy_id': 'claim_id'}, inplace=True)

    return policy_table, vehicles_table, customers_table, claims_table

def upload_df_to_s3():
    try:
        with suppress_stderr():
            # Install kaggle
            subprocess.run([sys.executable, "-m", "pip", "install", "kaggle", "--target", "/home/system_normal", "-i", "https://pypi.tuna.tsinghua.edu.cn/simple"], stderr=subprocess.DEVNULL)
            sys.path.append('/home/system_normal')
            from kaggle.api.kaggle_api_extended import KaggleApi

            # Setup Kaggle API
            config_dir = '/home/system_normal/tempdata/.config/kaggle'
            if not os.path.exists(config_dir):
                os.makedirs(config_dir)

            kaggle_json = {
                "username": "${kaggle_username}",
                "key": "${kaggel_key}"
            }
            with open(os.path.join(config_dir, 'kaggle.json'), 'w') as f:
                json.dump(kaggle_json, f)

            os.environ['KAGGLE_CONFIG_DIR'] = config_dir
            api = KaggleApi()
            api.authenticate()

            dataset = 'litvinenko630/insurance-claims'
            csv_file = 'Insurance claims data.csv'

            api.dataset_download_files(dataset, path='/home/system_normal/tempdata')

            with zipfile.ZipFile('/home/system_normal/tempdata/insurance-claims.zip', 'r') as zip_ref:
                zip_ref.extract(csv_file, path='/home/system_normal/tempdata')

            policy_data, vehicles_data, customers_data, claims_data = load_random_sample(f'/home/system_normal/tempdata/{csv_file}', 20)
            policy = policy_data.to_csv(index=False)
            vehicles = vehicles_data.to_csv(index=False)
            customers = customers_data.to_csv(index=False)
            claims = claims_data.to_csv(index=False)

            current_datetime = datetime.now().strftime("%Y%m%d_%H%M%S")
            
            s3_client = boto3.client(
                's3',
                aws_access_key_id= '${aws_access_key_id}',
                aws_secret_access_key= '${aws_secret_access_key}',
                region_name= '${aws_region_name}'
            )

            s3_bucket = 'insurance-data-clickzetta-etl-project'
            s3_key_policy = f'policy/policy_{current_datetime}.csv'
            s3_key_vehicles = f'vehicles/vehicles_{current_datetime}.csv'
            s3_key_customers = f'customers/customers_{current_datetime}.csv'
            s3_key_claims = f'claims/claims_{current_datetime}.csv'

            s3_client.put_object(Bucket=s3_bucket, Key=s3_key_policy, Body=policy)
            s3_client.put_object(Bucket=s3_bucket, Key=s3_key_vehicles, Body=vehicles)
            s3_client.put_object(Bucket=s3_bucket, Key=s3_key_customers, Body=customers)
            s3_client.put_object(Bucket=s3_bucket, Key=s3_key_claims, Body=claims)
            
    except Exception as e:
        print(f"An error occurred: {e}")

# Example usage:
upload_df_to_s3()

```

--------------------------------

### Load Random Sample from CSV in Python

Source: https://yunqi.tech/documents/MigrateSnowflakeRealtimeETLPipelinetoClickzettaLakehouse

This Python function reads a CSV file, selects a random sample of rows, and separates the data into distinct pandas DataFrames for policy, vehicles, customers, and claims. It's designed for efficient sampling without loading the entire file into memory if not needed.

```python
import pandas as pd
import random

def load_random_sample(csv_file, sample_size):

    # Count total rows in the CSV file
    total_rows = sum(1 for line in open(csv_file, encoding='utf-8')) - 1  # Subtract header row

    # Calculate indices of rows to skip (non-selected)
    skip_indices = random.sample(range(1, total_rows + 1), total_rows - sample_size)

    # Load DataFrame with random sample of rows
    df = pd.read_csv(csv_file, skiprows=skip_indices)

    policy_table = df[['policy_id', 'subscription_length', 'region_code', 'segment']].copy()
    vehicles_table = df[['policy_id', 'vehicle_age', 'fuel_type', 'is_parking_sensors', 'is_parking_camera', 'rear_brakes_type', 'displacement', 'transmission_type', 'steering_type', 'turning_radius', 'gross_weight', 'is_front_fog_lights', 'is_rear_window_wiper', 'is_rear_window_washer', 'is_rear_window_defogger', 'is_brake_assist', 'is_central_locking', 'is_power_steering', 'is_day_night_rear_view_mirror', 'is_speed_alert', 'ncap_rating']].copy()
    customers_table = df[['policy_id', 'customer_age', 'region_density']].copy()
    claims_table = df[['policy_id', 'claim_status']].copy()

    vehicles_table.rename(columns={'policy_id': 'vehicle_id'}, inplace=True)
    customers_table.rename(columns={'policy_id': 'customer_id'}, inplace=True)
    claims_table.rename(columns={'policy_id': 'claim_id'}, inplace=True)

    return policy_table, vehicles_table, customers_table, claims_table
```

--------------------------------

### Create Final Transformed Data Tables

Source: https://yunqi.tech/documents/MigrateSnowflakeRealtimeETLPipelinetoClickzettaLakehouse

SQL statements to create final tables for storing transformed data. These tables are designed to hold the processed information from stream objects.

```sql
CREATE  TABLE IF NOT EXISTS policy(
    policy_id VARCHAR(15) ,
    subscription_length FLOAT,
    region_code VARCHAR(5),
    segment VARCHAR(10));


CREATE  TABLE IF NOT EXISTS vehicles (
    vehicle_id VARCHAR(15) ,
    vehicle_age FLOAT,
    fuel_type VARCHAR(10),
    parking_sensors VARCHAR(5),
    parking_camera VARCHAR(5),
    rear_brakes_type VARCHAR(10),
    displacement INT,
    trasmission_type VARCHAR(20),
    steering_type VARCHAR(15),
    turning_radius FLOAT,
    gross_weight INT,
    front_fog_lights VARCHAR(5),
    rear_window_wiper VARCHAR(5),
    rear_window_washer VARCHAR(5),
    rear_window_defogger VARCHAR(5),
    brake_assist VARCHAR(5),
    central_locking VARCHAR(5),
    power_steering VARCHAR(5),
    day_night_rear_view_mirror VARCHAR(5),
    is_speed_alert VARCHAR(5),
    ncap_rating INT);

CREATE  TABLE IF NOT EXISTS customers(
    customer_id VARCHAR(20) ,
    customer_age INT,
    region_density INT);

CREATE  TABLE IF NOT EXISTS claims(
    claim_id VARCHAR(20) ,
    claim_status INT);
```

--------------------------------

### Policy Table Management Operations

Source: https://yunqi.tech/documents/MigrateSnowflakeRealtimeETLPipelinetoClickzettaLakehouse

SQL commands for managing the 'policy' table, including selecting all data, and truncating the table. These operations are useful for data inspection and resetting the table.

```sql
SELECT * FROM policy;
TRUNCATE TABLE policy;
```

--------------------------------

### Suppress Stderr Context Manager in Python

Source: https://yunqi.tech/documents/MigrateSnowflakeRealtimeETLPipelinetoClickzettaLakehouse

A Python context manager to temporarily redirect stderr to /dev/null, useful for silencing noisy subprocesses or library outputs during execution. It ensures stderr is restored even if errors occur.

```python
import sys
import os
import contextlib

@contextlib.contextmanager
def suppress_stderr():
    with open(os.devnull, 'w') as devnull:
        old_stderr = sys.stderr
        sys.stderr = devnull
        try:
            yield
        finally:
            sys.stderr = old_stderr
```
Tokens:
Raw