Data Engineer
dict1 = { (1, 2), [3, 4, 5] : 0 } var = 1, 2
МТСIn what situation can dictionary search degrade to the worst case?
ОстровокHow did you solve optimization problems when a query runs slowly and performs a full scan? How would you approach a new such task?
How to add a moving average of the order sum for the current and two previous days per user?
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
What does the depends_on_past parameter do in Airflow?
ОстровокWhat was the purpose of Data Vault for you in general?
What experience do you have with SQL and relational databases?
Has the method of comparing hashes of records been used to calculate the delta between the source and target tables?
What happens when data quality constraints are triggered: alert, crash, restart?
Лига Цифровой ЭкономикиHow did you work with 1C: did you retrieve data directly from the database, through a bus, or by another method?
Лига Цифровой ЭкономикиWhat are the disadvantages of UDFs in Spark and what is their downside?
ОстровокHave you configured tags in S3 on buckets or objects?
Лига Цифровой ЭкономикиWhat databases have you worked with? How is MongoDB structured and how does it scale?
How do you determine that a record already exists and does not need to be recorded again?
Лига Цифровой ЭкономикиNULL plus 5 — what will it be?
Is it possible to read data in parallel from a single Parquet file in Spark, or only with one core in one task?
Why can data be lost? Provide some reasons.
ОстровокWhat are lazy operations in Spark and what are they used for?
AstonTell about an interesting task in the last 2.5 years in the company, for example related to ClickHouse.