Data Engineer

Kokiais tikslais naudojamos lango funkcijos?

Kokia yra apribojimas naudojant DISTINCT ON PostgreSQL su ORDER BY?

Gavai užduotį su numeriu ABC. Koks tavo veiksmų tvarkas Git pradžioje?

Middle

Альфа-Банк

Ar kada nors dirbote su dekoratoriais Python? Kas jie yra ir kur jie taikomi?

Middle

Островок

Kokia yra shuffle pavojus Spark'e?

Middle

Островок

def extract_from_s3(**kwargs): df = pd.read_csv("s3://mano-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

Ką gali papasakoti apie SOAP API?

Middle+

Лига Цифровой Экономики

Kuo skiriasi ROWS BETWEEN ir RANGE BETWEEN?

Kaip pandas'e taikyti funkciją visiems eilutėms ar elementams?

Į ką reikėtų atkreipti dėmesį analizuojant didelius XML failus, kalbant apie atmintį, greitį ir tikslumą?

Kaip tiksliai skaitėte Parquet?

Ar reikėjo dirbti tiesiogiai su Spark? Kuo jis neigiamas?

Middle

МВидео

Kokiuose Scala projektuose dirbote anksčiau? Pasidalinkite konkrečiomis užduotimis ir pasiekimais šioje technologijoje.

Kaip parašei Airflow DAG ir užduotis: rankiniu būdu ar naudojant šablonus?

Middle

МВидео

Ataskaita logistikos įmonei Jūs esate logistikos įmonės analitikas, kuris registruoja operacijas sandėliuose. Turite paruošti ataskaitą apie kiekvieno sandėlio efektyvumą. Kiekvienam sandėliui apskaičiuokite: • bendrą operacijų skaičių (count_operations); • bendrą apdorotų prekių kiekį sandėlyje (sum_quantity); • vidutinį operacijos apdorojimo laiką (avg_processing_time), apskaičiuotą tik iš operacijų su nurodytu laiku (ne NULL), suapvalintą iki artimiausio sveiko skaičiaus; • didžiausią ir mažiausią prekių kiekį, apdorotą vienoje operacijoje (max_quantity, min_quantity); • kiekvieno tipo operacijų skaičių («pateikimas», «siuntimas», «perdavimas») atskirose stulpeliuose: supply_operations, shipment_operations, transfer_operations. Filtruokite sandėlius, kurių bendras operacijų skaičius yra daugiau nei 2 ir vidutinis apdorojimo laikas neviršija 60 minučių. Rodyti rezultatą pagal sandėlio ID didėjimo tvarka. Įvesties formatas Operacijų lentelė: • operation_id (int) — unikalus operacijos ID • warehouse_id (int) — sandėlio ID • operation_type (text) — operacijos tipas: «pateikimas», «siuntimas», «perdavimas»

Ar buvo atvejų, kai buvo sąveikos su reliacinėmis duomenų bazėmis?

Kas yra semantinis versijavimas? Kada didinti major, minor, patch?

Su priklausomybėmis tarp darbų, tarp DAG-ų — ar naudojote jutiklius, kad jie vienas po kito būtų paleisti?

Middle

МВидео

Kaip organizuojate savo projektus? Rašote README ar ką nors kitą?

Kada yra tinkamas laikas naudoti ThreadPoolExecutor?