Data Engineer

Ինչպես վարվել, երբ user_id-ի վրա կա skew (շեղում)՝ երբ միանում եք գործարքների և օգտվողների աղյուսակներին: Ինչպե՞ս ընտրել օպտիմալ բաշխման բանալին:

Ձախ կողմի միացում: աղյուսակ, որը ունի 10 գրանցում, Ձախ կողմի միացում՝ աղյուսակ, որը ունի 100 գրանցում: Ինչպիսի՞ն կարող է լինել ստացվող տողերի նվազագույն և առավելագույն քանակը։

միացեք clickhouse_driver from airflow.hooks.base import * def get_clickhouse_client(): conn = BaseHook.get_connection("clickhouse_default") return clickhouse_driver.Client( host=conn.host, port=conn.port, user=conn.login, password=conn.password, database=conn.schema )

Ի՞նչ է բաժանումը և բաշխումը (sharding):

CREATE TABLE core.localUserMetadata ON CLUSTER cluster_4x2 ( UserId String, Country String, LastLoginDate DateTime ) ENGINE = ReplikatsiyalashganReplacingMergeTree() ORDER BY (UserId); CREATE TABLE core.userMetadata ON CLUSTER cluster_4x2 ( UserId String, Country String, LastLoginDate DateTime ) ENGINE = Tarqatilgan('cluster_4x2', 'core', 'localUserMetadata', cityHash64(LastLoginDate));

interview.utils-dən get_clickhouse_client import edin airflow-dən DAG import edin airflow.operators.python-dən PythonOperator import edin airflow.sensors.external_task-dən ExternalTaskSensor import edin datetime-dən datetime import edin pandas-ı pd kimi import edin clickhouse_driver-ı import edin os-ı import edin CLICKHOUSE_CLIENT = get_clickhouse_client() default_args = { "start_date": datetime(2024, 1, 1) } with DAG( dag_id="datamarts.daily_revenue_per_country", default_args=default_args, schedule_interval="@daily", catchup=False ) as dag: 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" ) 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( ... )

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

Որ ռեսուրսն է առավել շատ սպառվում Nested Loop Join-ում?

ստեղծեք raw.local_transactions աղյուսակը cluster_4x2 խմբում ( transaction_id String, user_id String, amount Float64, created_at DateTime ) շարժիչ = ReplicatedMergeTree() բաժանարար ըստ amount վերևում ըստ (transaction_id); ստեղծեք raw.transactions աղյուսակը cluster_4x2 խմբում ( transaction_id String, user_id String, amount Float64, created_at DateTime ) շարժիչ = Distributed('cluster_4x2', 'raw', 'local_transactions', cityHash64(user_id));

CREATE TABLE datamarts.daily_revenue_per_country ON CLUSTER cluster_4x2 ( event_date Date, country String, total_revenue Float64 ) ENGINE = MergeTree() PARTITION BY toYYYYMM(event_date) ORDER BY (event_date);

Ի՞նչն է տարբերությունը ROWS BETWEEN և RANGE BETWEEN-ի միջև։

Որ ֆիզիկական JOIN ալգորիթմները աշխատում են տվյալների բազաներում?

Վահանակի ֆունկցիա SUM() OVER (PARTITION BY user_id) — մի դեպքում ավելացնում ենք ORDER BY գնելու ամսաթիվը, մյուս դեպքում՝ ոչ: Ինչ տարբերություն կա?

NULL plus 5 — ինչքա՞ն կլինի։