Data Engineer
Opowiedz o sobie: doświadczenia, zadania, funkcje lub osiągnięcia, z których jesteś dumny.
Czy masz doświadczenie z FastAPI?
AstonIle łącznie pracowałeś z Spark i jak głęboko się w niego zaangażowałeś?
dict1 = { (1, 2), [3, 4, 5] : 0 } var = 1, 2
МТСOpowiedz o interesującym zadaniu w firmie w ciągu ostatnich 2,5 roku, na przykład związanym z ClickHouse.
W jakiej sytuacji wyszukiwanie w słowniku może się pogorszyć do najgorszego przypadku?
ОстровокJak rozwiązywałeś problemy optymalizacyjne, gdy zapytanie działa długo i wykonuje pełny skan? Jak podszedłbyś do nowego takiego zadania?
Jak dodać ruchomą średnią sumy zamówień za bieżący i dwa poprzednie dni na użytkownika?
def extract_from_s3(**kwargs): df = pd.read_csv("s3://my-bucket/data/transactions_{}.csv".format(datetime.now().strftime("%Y-%m-%d"))) kwargs["ti"].xcom_push(key="df", value=df.to_dict()) def load_to_raw_table(**kwargs): df = pd.DataFrame(kwargs["ti"].xcom_pull(task_ids="extract", key="df")) rows = [tuple(r) for r in df[["transaction_id", "user_id", "amount", "created_at"]].to_numpy()] CLICKHOUSE_CLIENT.execute( "INSERT INTO raw.transactions (transaction_id, user_id, amount, created_at) VALUES", rows ) def build_aggregate_view(): query = """ INSERT INTO datamarts.daily_revenue_per_country SELECT toDate(r.created_at) as event_date, u.country, sum(r.amount) as total_revenue FROM raw.transactions r LEFT JOIN core.userMetadata u ON r.user_id = u.UserId WHERE toDate(r.created_at) = '{}' GROUP BY event_date, u.country """.format(datetime.now().strftime("%Y-%m-%d")) CLICKHOUSE_CLIENT.execute(query) transactions_sensor = S3KeySensor( task_id="transactions_sensor", bucket_key="data/transactions_{}.csv".format(datetime.now().strftime("%Y-%m-%d")), bucket_name="my-bucket", aws_conn_id="aws_default", timeout=600, poke_interval=30, mode="poke" ) extract = PythonOperator( task_id="extract", python_callable=extract_from_s3, provide_context=True ) load = PythonOperator( task_id="load", python_callable=load_to_raw_table, provide_context=True ) aggregate = PythonOperator( task_id="aggregate", python_callable=build_aggregate_view ) transactions_sensor >> extract >> load >> aggregate
Co robi parametr depends_on_past w Airflow?
ОстровокDo czego w ogóle potrzebowałeś Data Vault?
Opowiedz o zadaniu związanym z konfiguracją replikacji, które rozwiązałeś.
Czy użyto metody porównywania hashy rekordów do obliczenia różnicy między źródłem a tabelą docelową?
Co się działo, gdy uruchamiano ograniczenie jakości danych: alert, awaria, ponowne uruchomienie?
Лига Цифровой ЭкономикиPorównaj podejścia ETL i ELT: jaka jest różnica i kiedy każde z nich jest stosowane?
Jak pracowali z 1C: pobierali dane bezpośrednio z bazy danych, przez magistralę lub w inny sposób?
Лига Цифровой ЭкономикиZ jakimi bazami danych pracowałeś? Jak działa MongoDB i jak się skaluje?
Jak ustalili, że wpis już istnieje i nie trzeba go ponownie zapisywać?
Лига Цифровой ЭкономикиNULL plus 5 — ile to będzie?
Czy można odczytywać dane równolegle z jednego pliku Parquet w Spark, czy tylko jednym rdzeniem w jednym zadaniu?