Data Engineer
Fale sobre si: experiência, tarefas, recursos ou conquistas de que se orgulhe.
Tem experiência com FastAPI?
AstonQuanto tempo trabalhou com Spark no total e quão profundamente se envolveu nele?
dict1 = { (1, 2), [3, 4, 5] : 0 } var = 1, 2
МТСFale sobre uma tarefa interessante na empresa nos últimos 2,5 anos, por exemplo, relacionada com o ClickHouse.
Em que situação a busca por um dicionário pode degradar-se até ao pior caso?
ОстровокComo resolveu problemas de otimização quando a consulta demora muito e realiza uma varredura completa? Como abordaria uma nova tarefa semelhante?
Como adicionar uma média móvel da soma dos pedidos do dia atual e dos dois dias anteriores por usuário?
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
O que faz o parâmetro depends_on_past no Airflow?
ОстровокPara que é que você precisava do Data Vault, em geral?
Fale sobre a tarefa de configuração de replicação que você resolveu.
Foi utilizado o método de comparação de hashes de registos para calcular a diferença entre a fonte e a tabela de destino?
O que acontecerá ao aplicar `s ** 2` à lista `s = [1, 2, 3]`?
O que acontecia quando a restrição de qualidade dos dados era acionada: alerta, queda, reinício?
Лига Цифровой ЭкономикиCompare as abordagens ETL e ELT: qual é a diferença e quando cada uma delas é aplicada?
Como trabalhavam com 1C: extraíam dados diretamente do banco de dados, através de um barramento ou de outra forma?
Лига Цифровой ЭкономикиCom que bases de dados trabalhou? Como funciona o MongoDB e como ela escala?
Como determinaram que a entrada já existe e que não é necessário gravá-la novamente?
Лига Цифровой ЭкономикиNULL mais 5 — quanto será?