مهندسی داده با DLT و REST

Summarize this content to 400 words in Persian Lang

ساخت خط لوله داده با DLT

در این مثال، یک شرکت داده های IoT را از طریق یک API تقریباً هم زمان برای نظارت و یک خوراک API تاریخی برای تجزیه و تحلیل ارائه می دهد. هر کدام هدف خاصی را دنبال می کنند، اما می توانید از یک راه حل ETL (استخراج/تبدیل/بارگذاری) برای مدیریت داده ها استفاده کنید. API ها روشی مناسب برای دسترسی به داده ها از وسایل نقلیه و دستگاه ها هستند. با این حال، یک نقطه پایانی ممکن است بیش از حد مجاز باشد. یکی از روش‌های مقیاس‌سازی، وارد کردن داده‌ها در یک پایگاه داده توزیع‌شده است. این مقاله نشان می دهد که چگونه داده ها را از یک REST API به DuckDB، یک پایگاه داده توزیع شده که برای پردازش تحلیلی آنلاین (OLAP) و تجزیه و تحلیل داده های تاریخی ساخته شده است، وارد کنید.

الزامات

برای ساخت این راه حل، شما نیاز دارید:

پایتون 3.10 یا بالاتر

duckDB به صورت محلی نصب شده است

راه اندازی

ایجاد یک محیط اختصاصی برای پروژه پایتون تمرین خوبی است. یک دایرکتوری و یک محیط ایجاد کنید و محیط را فعال کنید.

mkdir rest_iot
cd rest_iot
python3 -m venv venv
source venv/bin/activate

وارد حالت تمام صفحه شوید

از حالت تمام صفحه خارج شوید

این مثال از data load tool – dlt برای مصرف و مصرف داده های تقریباً هم زمان از API. Dlt یک کتابخانه سبک وزن آماده استخراج، تبدیل و بارگذاری (ETL) پایتون است. برای نصب dbt:

pip install -U dlt[duckdb]”

وارد حالت تمام صفحه شوید

از حالت تمام صفحه خارج شوید

می توانیم از dbt برای ایجاد یک پروژه جدید استفاده کنیم.

dlt init iot duckdb

وارد حالت تمام صفحه شوید

از حالت تمام صفحه خارج شوید

Dbt چندین فایل پروژه ایجاد می کند، از جمله یک مثال کد، requires.txt، و یک فهرست با فایل های پیکربندی.

iot_pipeline.py
requirements.txt
.dlt/
config.toml
secrets.toml

وارد حالت تمام صفحه شوید

از حالت تمام صفحه خارج شوید

را requirements.txt فایل حاوی لیستی از بسته های پایتون مورد نیاز برای پروژه است. نصب این بسته ها برای اطمینان از اجرای روان پروژه ضروری است.

pip install -r requirements.txt

وارد حالت تمام صفحه شوید

از حالت تمام صفحه خارج شوید

مرحله آخر پیکربندی کلید API است. کلید API خود را به secrets.toml فایل

rest_token = “rest_api_aaaaaaaaaaaaabbbbbbbbbbbbbbbbccccccccc”

وارد حالت تمام صفحه شوید

از حالت تمام صفحه خارج شوید

داده های نزدیک به زمان واقعی

ما از کد نمونه تولید شده توسط استفاده نخواهیم کرد dlt. فایل را باز کنید و کد را پاک کنید. کد زیر را کپی و پیست کنید.

import dlt
from dlt.sources.rest_api import rest_api_source
from dlt.sources.helpers.rest_client.auth import BearerTokenAuth

nrt_source = rest_api_source({
“client”: {
“base_url”: “https://api.rest-iot.com/fleet/vehicles/stats/”,
“auth”: BearerTokenAuth(token=dlt.secrets[“rest_token”]),
“paginator”: {
“type”: “json_response”,
“next_url_path”: “paging.next”,
},
},
“resources”: [
{
“name” : “feed”,
“endpoint” : {
“params”: {
“types”: “gps”,
“decorations”: “obdOdometerMeters”
},
}
},
],
})

pipeline = dlt.pipeline(
pipeline_name=”feed_pipeline”,
destination=”duckdb”,
dataset_name=”iot”,
progress=”log”,
)

load_info = pipeline.run(nrt_source)

وارد حالت تمام صفحه شوید

از حالت تمام صفحه خارج شوید

کد منبعی را تعریف می کند که شامل نقطه پایانی REST IoT و نشانه API است. در منبع، منبع نزدیک به زمان واقعی همان است gps، که در duckdb بارگذاری می شود. Dlt صفحه‌بندی را به‌طور خودکار انجام می‌دهد و داده‌ها را تا زمانی که دیگر صفحه‌ای وجود نداشته باشد استخراج می‌کند.

قسمت دوم کد یک خط لوله برای استخراج داده ها ایجاد می کند. duckdb را به عنوان مقصد تعریف می کند و داده ها را در پایگاه داده iot می نویسد.

برای استخراج و بارگذاری داده ها، موارد زیر را اجرا کنید:

python3 iot_pipeline.py

وارد حالت تمام صفحه شوید

از حالت تمام صفحه خارج شوید

خط لوله هنگام اجرای برنامه، گزارش را به کنسول خروجی می دهد. ما می توانیم خروجی را با پرس و جو از duckdb با CLI بررسی کنیم.

duckdb

v1.1.3 19864453f7
Enter “.help” for usage hints.
Connected to a transient in-memory database.
Use “.open FILENAME” to reopen a persistent database.

وارد حالت تمام صفحه شوید

از حالت تمام صفحه خارج شوید

فایل پایگاه داده را در duckdb CLI باز کنید. برای پرس و جو از پایگاه داده می توانید از SQL استفاده کنید.

.open iot.duckdb
SHOW TABLES;

وارد حالت تمام صفحه شوید

از حالت تمام صفحه خارج شوید

SELECT * FROM feed;

وارد حالت تمام صفحه شوید

از حالت تمام صفحه خارج شوید

SELECT * FROM feed_gps;

وارد حالت تمام صفحه شوید

از حالت تمام صفحه خارج شوید

این مثال از duckdb استفاده می کند، اما dlt از بسیاری از مقاصد، از جمله Clickhouse، Snowflake، Databricks، S3 و بیش از 30 پایگاه داده SQL پشتیبانی می کند. توسعه‌دهندگان می‌توانند باطن داده‌ای را انتخاب کنند که نیازهای آنها را برآورده کند.

داده های تاریخی

بازیابی داده های تاریخی از همین الگو پیروی می کند، یعنی یک منبع را تعریف کرده و یک خط لوله ایجاد کنید.

params = {
“types”: “engineStates”,
“startTime”: “2020-07-23T00:00:00Z”,
“endTime”: “2020-07-24T00:00:00Z”
}

historical_source = rest_api_source({
“client”: {
“base_url”: “https://api.rest-iot.com/fleet/vehicles/stats/”,
“auth”: BearerTokenAuth(token=dlt.secrets[“rest_token”]),
“paginator”: {
“type”: “json_response”,
“next_url_path”: “paging.next”,
},
},
“resources”: [
{
“name” : “history”,
“endpoint” : {
“params”: params,
}
},
],
})

pipeline = dlt.pipeline(
pipeline_name=”history_pipeline”,
destination=”duckdb”,
dataset_name=”iot”,
progress=”log”,
)

load_info = pipeline.run(historical_source)

وارد حالت تمام صفحه شوید

از حالت تمام صفحه خارج شوید

شما می توانید به صورت برنامه نویسی با داده ها در duckdb کار کنید، اما تحلیلگران اغلب از صفحات گسترده استفاده می کنند. صادر کردن داده ها به یک فایل CSV ساده است. در جلسه duckdb از COPY دستور نوشتن داده ها در یک فایل

COPY history TO ‘history.csv’ (HEADER, DELIMITER ‘,’);

وارد حالت تمام صفحه شوید

از حالت تمام صفحه خارج شوید

چه چیزی خواهید ساخت؟

این مقاله نحوه کار با داده های تقریباً واقعی و تاریخی را با استفاده از این نشان می دهد dlt بسته بندی چه نیاز به مقیاس دسترسی به داده ها در سراسر سازمان داشته باشید یا داده های تاریخی را برای تجزیه و تحلیل پس از رویداد ارائه دهید، می توانید از همان چارچوب برای ارائه داده های مشتری استفاده کنید. در مقاله آینده، نحوه استفاده را نشان خواهم داد dlt با یک ارکستراتور جریان کار مانند Apache Airflow یا Dagster.“

mkdir rest_iot 
cd rest_iot
python3 -m venv venv
source venv/bin/activate

pip install -U dlt[duckdb]"

dlt init iot duckdb

iot_pipeline.py
requirements.txt
.dlt/
 config.toml
 secrets.toml

pip install -r requirements.txt

rest_token = "rest_api_aaaaaaaaaaaaabbbbbbbbbbbbbbbbccccccccc"

import dlt
from dlt.sources.rest_api import rest_api_source
from dlt.sources.helpers.rest_client.auth import BearerTokenAuth

nrt_source = rest_api_source({
    "client": {
        "base_url": "https://api.rest-iot.com/fleet/vehicles/stats/",
        "auth": BearerTokenAuth(token=dlt.secrets["rest_token"]),
        "paginator": {
            "type": "json_response",
            "next_url_path": "paging.next",
        },
    },
    "resources": [
        { 
            "name" : "feed",
            "endpoint" : {
                "params": {
                    "types": "gps",
                    "decorations": "obdOdometerMeters"
                },
            }
        },
    ],
})

pipeline = dlt.pipeline(
    pipeline_name="feed_pipeline",
    destination="duckdb",
    dataset_name="iot",
    progress="log",
)

load_info = pipeline.run(nrt_source)

python3 iot_pipeline.py

duckdb

v1.1.3 19864453f7
Enter ".help" for usage hints.
Connected to a transient in-memory database.
Use ".open FILENAME" to reopen a persistent database.

.open iot.duckdb
SHOW TABLES;

SELECT * FROM feed;

SELECT * FROM feed_gps;

params = {
    "types": "engineStates",
    "startTime": "2020-07-23T00:00:00Z",
    "endTime": "2020-07-24T00:00:00Z"
}

historical_source = rest_api_source({
    "client": {
        "base_url": "https://api.rest-iot.com/fleet/vehicles/stats/",
        "auth": BearerTokenAuth(token=dlt.secrets["rest_token"]),
        "paginator": {
            "type": "json_response",
            "next_url_path": "paging.next",
        },
    },
    "resources": [
        { 
            "name" : "history",
            "endpoint" : {
                "params": params,
            }
        },
    ],
})

pipeline = dlt.pipeline(
    pipeline_name="history_pipeline",
    destination="duckdb",
    dataset_name="iot",
    progress="log",
)

load_info = pipeline.run(historical_source)

COPY history TO 'history.csv' (HEADER, DELIMITER ',');