Integracja AWS - pobieranie wszystkich tabel Tulip w funkcji Lambda

Usprawnienie pobierania danych z Tulip do AWS dla szerszych możliwości analitycznych i integracyjnych

Cel

W tym przewodniku opisano krok po kroku, jak pobrać wszystkie dane z tabel Tulip do AWS za pomocą funkcji Lambda.

Wykracza to poza podstawowe zapytanie pobierające i iteruje przez wszystkie tabele w danej instancji; może to być świetne rozwiązanie dla cotygodniowego zadania ETL (Extract, Transform, Load)

Funkcja lambda może być wyzwalana za pomocą różnych zasobów, takich jak timery Event Bridge lub API Gateway

Poniżej przedstawiono przykładową architekturę:

Konfiguracja

Ta przykładowa integracja wymaga następujących elementów:

• Wykorzystanie API Tulip Tables (Uzyskaj klucz API i sekret w Ustawieniach konta)
• Tabela Tulip (Uzyskaj unikalny identyfikator tabeli)

Kroki wysokiego poziomu: 1. Utworzenie funkcji AWS Lambda z odpowiednim wyzwalaczem (API Gateway, Event Bridge Timer itp.) 2. Pobierz dane z tabeli Tulip za pomocą poniższego przykładu ``python import json import pandas as pd import requests from sqlalchemy import create_engine import os

def lambda_handler(event, context): # initialize db host = os.getenv('host') user = os.getenv('username') password = os.getenv('password') db = os.getenv('database')

engine_str = f'postgresql://{user}:{password}@{host}/{db}' engine = create_engine(engine_str) api_header = {'Authorization' : os.getenv('tulip_api_basic_auth')} instance = os.getenv('tulip_instance') base_url = f'https://{instance}.tulip.co/api/v3' get_tables_function = '/tables' r = requests.get(base_url+get_tables_function, headers=api_header) df = pd.DataFrame(r.json()) # funkcja konwertująca słowniki na ciągi def dict_to_str(cell): if isinstance(cell, dict): return str(cell) return cell # funkcja query_table def query_table(table_id, base_url, api_header): offset = 0 function = f'/tables/{table_id}/records?limit=100&offset={offset}&includeTotalCount=false&filterAggregator=all" r = requests.get(base_url+function, headers=api_header) df = pd.DataFrame(r.json()) length = len(r.json()) while length > 0: offset += 100 function = f'/tables/{table_id}/records?limit=100&offset={offset}&includeTotalCount=false&filterAggregator=all' r = requests.get(base_url+function, headers=api_header) length = len(r.json()) df_append = pd.DataFrame(r.json()) df = pd.concat([df, df_append], axis=0) df = df.apply(lambda row: row.apply(dict_to_str), axis=1) return df # create function def write_to_db(row, base_url, api_header): table = row['label'] id = row['id'] df = query_table(id, base_url, api_header) df.to_sql(table,engine, if_exists='replace', index=False) print(f'wrote {table} to database!') # iterate through all tables: df.apply(lambda row: write_to_db(row, base_url, api_header), axis=1) return { 'statusCode': 200, 'body': json.dumps('wrote to db!')}

## Przypadki użycia i kolejne kroki


Po sfinalizowaniu integracji z lambdą można łatwo analizować dane za pomocą notatnika sagemaker, QuickSight lub wielu innych narzędzi.


**1. Przewidywanie usterek**- Identyfikacja usterek produkcyjnych przed ich wystąpieniem i zwiększenie liczby poprawek za pierwszym razem - Identyfikacja głównych czynników wpływających na jakość produkcji w celu wdrożenia ulepszeń.


**2. Optymalizacja kosztów jakości**- identyfikacja możliwości optymalizacji projektu produktu bez wpływu na zadowolenie klienta.


**3. Optymalizacja zużycia energii**w produkcji - identyfikacja dźwigni produkcyjnych w celu optymalizacji zużycia energii.


**4. Przewidywanie i optymalizacja dostaw i planowania**- optymalizacja harmonogramu produkcji w oparciu o zapotrzebowanie klientów i harmonogram zamówień w czasie rzeczywistym.


**5. Globalna analiza porównawcza maszyn/linii**- analiza porównawcza podobnych maszyn lub urządzeń z normalizacją.


**6. Globalne / regionalne cyfrowe zarządzanie wydajnością**- skonsolidowane dane do tworzenia pulpitów nawigacyjnych w czasie rzeczywistym