​​Задачки 2️⃣ и 3️⃣

Дана таблица:

CREATE TABLE IF NOT EXISTS prices (
      id bigint not null,
      price int4 not null,
      dt date not null
  );

Пример данных:
id    price       dt
1     100   2021-01-01
1     110   2021-05-05

В таблице prices хранятся данные по ценам товаров. В этой таблице есть другие цены и другие товары, эти приведены для примера.
При изменении цены записывается новая строка с датой начиная с которой это цена действует.
Цена действует до тех пор, пока она не сменится другой ценой.
Надо написать параметризованный запрос, который по id и dt вернет цену товара на этот день.
Например:id=1, dt=2021-01-02 => price=100id=1, dt=2021-06-01 => price=110.

+ дополнительный вопрос от меня Какой механизм реализован в этой таблице?
——————————————————

Дана таблица Client (id, email, phone)
1. Найти все email которые встречаются в таблице более 2х раз.
2. Вывести id, email, email_дуплицирован(1,0)

☝️1 и 2 - два разных запроса

.
Жду ваших решений в комментариях и помните, у задачки необязательно есть одно единственное решение😉
#вопросы #собес #sql

​​Задача 4️⃣ (в двух частях)
.

Дана таблица:
CREATE TABLE IF NOT EXISTS deliveries (
      id bigint not null,
      dt date not null,
      id_supplier bigint not null,
      id_product bigint not null,
      quantity int4 not null,
      price decimal(16, 4)
  );

Пример данных:
id    dt    id_supplier       id_product  quantity    price
1     2021-01-01  2     134   5     10.5
2     2021-05-05  1     135   10    11.0

Задача 2.1
В таблице хранятся данные о поставках, каждая поставка имеет:
 свой уникальный id, дату совершения поставки dt, данные о поставщике id_supplier и товаре id_product, количестве штук товара в поставке quantity, цене одной штуки price.
 Необходимо найти поставщика, который поставил нам товаров на наибольшую сумму денег.

Задача 2.2

Дополнительно дан справочник:
CREATE TABLE IF NOT EXISTS supplier (
      id bigint not null,
      name VARCHAR(200) not null
  );

Пример данных:
id    name
1     Поставщик1
2     Поставщик2

Как найти поставщиков которые ничего не поставляли, если рядом есть справочник поставщиков?

.
Жду ваших верных решений😎
#вопросы #собес #sql

​​Задача 5️⃣
.
Продолжаем серию SQL задачек, на очереди аналитическая задачка.
.

Посчитать среднее время нахождения в статусе "Подтверждён" , за которым следовал статус "Доставлен" для заказов , созданных в этом году.

-- Таблица, со статусами заказа
create table OrderStatusChangeLog
(
      order_id          INTEGER 
    , status_id         INTEGER
    , time_changed      DATETIME
);
comment on column OrderStatusChangeLog.order_id is 'ID заказа';
comment on column OrderStatusChangeLog.status_id is 'ID статуса заказа';
comment on column OrderStatusChangeLog.time_changed is 'Время изменения заказа';

Пример (сэмпл) данных в таблице:
+--------+---------+-------------------+
|order_id|status_id|time_changed       |
+--------+---------+-------------------+
|14324   |1        |01.05.2022 10:00:01|
|14324   |2        |01.05.2022 10:25:17|
|14324   |10       |01.05.2022 10:30:55|
|14324   |7        |01.05.2022 10:31:24|
|14324   |10       |01.05.2022 15:30:54|
|14324   |5        |04.05.2022 12:22:32|
|1468    |1        |01.05.2022 10:00:01|
|1468    |2        |01.05.2022 11:25:23|
|1468    |10       |01.05.2022 11:30:48|
|67825   |1        |01.05.2022 10:00:01|
|67825   |2        |01.05.2022 10:25:44|
|67825   |10       |01.05.2022 10:30:05|
|67825   |5        |08.05.2022 16:32:49|
+--------+---------+-------------------+


-- Таблица-справочник статусов
create table OrderStatus
(
      status_id         INTEGER
    , status_name       VARCHAR(255)
);
comment on column OrderStatus.status_id is 'ID статуса заказа';
comment on column OrderStatus.status_name is 'Наименование статуса заказа';

Пример (сэмпл) данных в таблице:
+---------+-----------+
|status_id|status_name|
+---------+-----------+
|1        |Новый       |
|2        |Оплачен    |
|10       |Подтверждён|
+---------+-----------+


-- Таблица с заказами
create table Orders
(
      order_id          INTEGER PRIMARY KEY
    , time_created      DATETIME
    , client_id         INTEGER
    , status_id         INTEGER
);
comment on column Orders.order_id is 'ID заказа';
comment on column Orders.time_created is 'Время создания заказа';
comment on column Orders.client_id is 'ID клиента';
comment on column Orders.status_id is 'ID текущего статуса заказа';

Пример (сэмпл) данных в таблице:
+--------+-------------------+---------+---------+
|order_id|time_created       |client_id|status_id|
+--------+-------------------+---------+---------+
|1468    |01.05.2022 10:00:01|234      |10       |
|14324   |01.05.2022 10:00:01|738294   |5        |
|3421987 |10.03.2021 19:03:55|761285   |5        |
+--------+-------------------+---------+---------+

.
Дерзайте, задачку точно можно решить не одним способом😉 Используйте все ваши знания и творческий подход😎
#вопросы #собес #sql

​​Субботний мем
.
https://youtu.be/NowOKup33VQ

​​Задачки с собесов - Python
.
В работе дата инженера нередко приходится тянуть данные из API, в работе с ним помогает Python, поэтому есть и задачки по нему. Встречаем первую:

Ответ от API:
 
response = '''
{
    "trace_id": "d6518bbb-b47b-4d1b-931e-74013148872d",
    "type": "ADD",
    "data": [
        {
            "id": 464612,
            "timestamp": "2023-05-04T09:24:57Z",
            "is_rejected": false,
            "reason": null,
            "values": [13.4, 64.7, 12]
        },
        {
            "id": 464612,
            "timestamp": "2023-05-04T10:32:15Z",
            "is_rejected": false,
            "reason": null,
            "values": [77.3, 7.6, 10]
        },
        {
            "id": 74641123,
            "timestamp": "2023-05-04T05:23:13Z",
            "is_rejected": true,
            "reason": "Code 53",
            "values": [45.2, 0.5, 0]
        }
    ]
}
'''
 
В ответе от API не хватает поля name, его можно получить вызвав функцию get_name_by_id:
 
def get_name_by_id(id: int) -> str:
    from random import randrange
    from time import sleep
     
    sleep(1)
 
    box_number = randrange(0, 1000) // len(str(id))
    return f"Box {box_number}"

.
При решении внимательно приглядитесь к функции get_name_by_id, неспроста там есть некоторые моменты☝️
.
Пока решаете, вспомните с какими "чудесными" API вам приходилось работать, пишите их в комментах.
#вопросы #собес #python

А такое даже не грешно запостить: смотрите как изменилась картина тулзов для дата инженеров🙈

https://bar-or-pie.dianov.org/result-1300.html
Бар о пай

📈Это однозначно лучшее про датавиз-гигиену
.
Пройдите тест и узнайте:
- в каких случаях стоит использовать цветовую кодировку, а когда нет
- когда обводка это зло
- когда следует повернуть ваш барплот
- или когда не следует подписывать все числа на графике
.
И наконец ответьте себе на главный вопрос: Bar or Pie😉
ps: а еще там есть ссылки, в которых более подробно изложена та или иная идея
.
#plot #data #bar #pie

​​😎А вы уже еBASH.....
.
Не перестаю удивляться магии линукса и простых (но это не точно) консольных команд.
.
Разбираясь с уткой по-пекински 🦆 (про неё тоже будет пост, довольно интересная альтернатива sqlite3, особенно если хочется практиковать SQL) в доке обнаружил утилиту для построения графиков в терминале youplot 😈. В кратком tutorial можно найти довольно приличный набор графиков:
- столбчатые
- линейные
- гистограммы или графики плотности
- scatterplot
- boxplot

Кажется, для базового анализа выше крыши, так что теперь разведывательный анализ можно выполнять без Python😁
.
Конечно, чтобы построить график нужно не бояться терминала (вообще кажется, в нем можно сделать всё и немножечко больше).
.
Вот некоторые сниппеты:

<code class="language-bash"># данные из Python
echo -e "from numpy import random;" 
"n = random.randn(25000);"  
"print('\\n'.join(str(i) for i in n))" 
| python3 
| uplot hist --nbins 51

Данные из файла на github:

<code class="language-bash">curl -s https://raw.githubusercontent.com/urevoleg/example-direct-access/main/kinopoisk/kinopoisk_parsing.csv | awk -F';' '{print $20}' | uplot hist --nbins 20

В общем изучаем магию🧙‍♀️

А вы знаете что-то хитрое в linux-терминале?

#bash #duckdb #youplot

​​🔥По горячим следам
.
Случился стандартный рабочий кейс: надо затащить несколько внешних таблиц в клик.
Источник: Postgres
Сущность: N вьюх (у одной из под капотом было около десятка JOIN 🥶, дадим ей имя вьюха G)
История умалчивает подробности и даже почему выбран такой способ загрузки: целевая таблица полностью перегружается каждый день.
.
Работа шла стандартным флоу (практикую частичное решение задач):
- написал даг с коннектом к источнику
- таски для создания инфры
- несколько таск групп с небольшим селектом из источника (проверить, что всё ок и источник отдает данные)
- заварил 🫖 и стал налюдать за даграном
.
Всё шло хорошо до момента тестирования вьюхи G. В логах вижу инфра создается, как только дело доходит до забора данных, то таска ждет и падает по SIGTERM - явно что-то идет не так.
.
В процессе обсуждения с разработчиком со стороны источника выяснилось:
- про десятки JOIN
- там в таблице 10+млн строк
- .....
.
Но тут я подумал: ведь я же батчами забираю, причем здесь млн строк🤔.
.
Код для загрузки батчами тривиален:

<code class="language-python"><code class="language-python">import psycopg2

con = psycopg2.connect(.....)
...
for chunk in pd.read_sql(stmt, con=con, chunk_size=100500):
    yield chunk

.
Крутил, вертел никак, надо идти курить доку (тимлид подкинул про именованные курсоры). И от тут истина и открылась, если делать так как выше, то:
- stmt должно выполниться полностью на стороне сервера
- полные данные передаются на клиент
- и батчи получаются на стороне клиента - то есть у нас клиентский курсор
И естественно дождаться не получается пока собирутся 10+млн строк.
.
Из доки psycopg2 узнал про реализацию серверного курсора (тот который будет формировать батчи на стороне сервера и отдавать их). Для организиации серверного курсора меняем код, использовать красивую pandas-обертку уже не получится:

<code class="language-python">import psycopg2
from psycopg2.extras import RealDictCursor

with psycopg2.connect(...) as conn:
    with conn.cursor(name="It's server cursor", cursor_factory=RealDictCursor()) as cursor:
        cur.execute(stmt)
        chunk = cur.fetchmany(chunk_size)
        while chunk:
            yield pd.DataFrame(chunk)
            chunk = cur.fetchmany(chunk_size)

.
Бинго, теперь батчи отдаются по мере готовности и не нужно ждать сбора всех данных - пошло чтение данных. Далее я дописал загрузку в Clickhouse.
.
ps: на источнике окаалось 36+млн строк 🤨
#case #work #psycopg2 #def

​​Clickhouse-notes 📔
.
Буду делиться лайфхаками, которые использую в работе. Отмечу, что на работе клик распределенный и это накладывает некоторые ограничения или нюансы в работе ( а вы как хотели, только плюсы чтоль)
.
0️⃣Стоит отметить такой факт: клик распределенный -> нет мастер ноды -> координирует работу Zookeeper -> запросы можно выполнять как ON CLUSTER так и на каждом хосте отдельно
В первом подходе организуется distributed_ddl_queue, которая иногда очень долго рассасывается, при втором подходе мы обходим эту очередь, выполняя запросы на каждом хосте отдельно - это можно назвать лайфхаком номер 0
.
Операции DELETE\ UPDATE  дорогие для клика (это мутации, выпоняются через ALTER TABLE) - их вообще следует избегать, используя семейство движков MergeTree.  Но иногда без них нельзя, аналитикам только дай, что поделитить или поапдейтить ->  копиться очередь -> её надо обходить
.
Локально у меня всегда в docker запущен clickhouse (однонодовый) для экспериментов, с однонодовым в разы легче и проще, а скорость его работы, ну вы знаете 🚀
.
ps: где-то было видео со сравнением работы одного и того же запроса на Postgres и Clickhouse, найду, отправлю в комменты. Понятно что сравнение ежа со слоном, но когда я первый раз так перешел был приятно удивлен 🙃

#clickhouse #lifehacks

🚛 Как грузить в клик
.
Из ответа на вопрос в комментариях родился маленький пост.
Для загрузки данных в clickhouse в команде используем два подхода:
1️⃣ Pythonic way с использованием clickhouse_driver

Тут два метода:
- execute
- insert_dataframe

Кому, что больше нравиться, встречаются ошибки с типами данных

2️⃣ Native way с использованием linux клиента clickhouse-client
Например, способом ниже загружаются csv файлы

<code class="language-shell"><code class="language-bash">cat file.csv | clickhouse-client --host=$CH_HOST --user=$CH_USER --password=$CH_PASSWORD --query="INSERT INTO $CH_SCHEMA.$CH_TABLE FORMAT CSVWithNames"

Оба подхода используют TCP соединение.

Есть еще третий, с использованием HTTP, ниже пример bash скрипта:

<code class="language-bash">#!/bin/bash


#CH
CLUSTER_NAME=
CH_HOST=192.168.55.121
CH_PORT=8123
CH_SCHEMA=stage
CH_USER=clickadmin
CH_PASSWORD=aLpjr5HMq
CH_TABLE=TMP_CURL2CH
CH_URL="http://${CH_HOST}:${CH_PORT}"

DDL_QUERY="CREATE OR REPLACE TABLE $CH_SCHEMA.$CH_TABLE
(
    user_id String,
    name Nullable(String),
    email Nullable(String),
    phone Nullable(String),
    address Nullable(String),
    birth_date Nullable(Date)
)
ENGINE = MergeTree
ORDER BY (birth_date, name, phone)
SETTINGS index_granularity = 8192, allow_nullable_key=1;"

echo "☠️ DDL Executing...."
echo $DDL_QUERY | curl -H "X-ClickHouse-User: $CH_USER" -H "X-ClickHouse-Key: $CH_PASSWORD" "$CH_URL?query=" --data-binary @-


CSV_FILE="src/data.csv"

# Определение запроса на загрузку данных
INSERT_QUERY="INSERT+INTO+$CH_SCHEMA.$CH_TABLE+(user_id,name,email,phone,address,birth_date)+FORMAT+CSVWithNames"
echo "🚰 Inserting data from file: $CSV_FILE...."
echo "$CH_URL?query=$INSERT_QUERY"
cat $CSV_FILE | curl --data-binary @- -H "X-ClickHouse-User: $CH_USER" -H "X-ClickHouse-Key: $CH_PASSWORD" "$CH_URL?query=$INSERT_QUERY"

Вероятно, бывает иногда полезный, пока использовать не приходилось. У 3 способа есть особенности, о них лучше почитать в доке,например, у меня завелась передача user\pass только через заголовки.

#clickhouse #de #dataload #load

​​Clickhouse-notes⚡️
.
Предикат  ORDER BY при создании таблицы очень важная штука, при неверном выборе производительность запроса может быть крайне низкой.
.
Простыми словами: указанием полей в этой секции можно рассматривать как индекс, то есть сюда следует включать все столбцы, которые используются при фильтрации: это позволяет при чтении данных пропускать неподходящие куски -> то есть не читать лишнего (а иногда это ОЧЕНЬ много лишнего).
Если заранее неизвестно что и как будет фильтроваться, использую общее правило:

В секции ORDER BY указываю все поля в порядке увеличения кардинальности (то есть ближе в ORDER BY будет поле с наименьшим кол-вом уникальных значений)

Конечно не стоит увлекаться по все поля, но условно интуитивно понятные🙃
.
На чтение кол-ва данных также влияет партицирование, задаётся в секции PARTITION BY - об этом есть в доке.
.
Для того чтобы оценить эффективность ваших действий по управлению секциями ORDER BY\PARTITION BY использую перед запросом указание explain estimate, результат показывает сколько партов и строк будет прочитано в результате запроса, пример ниже:

<code class="language-markdown">
database|table  |parts|rows|marks|
--------+-------+-----+----+-----+
stg     |tmp_bpr|    1| 210|    1|

.
Подбираю так, чтобы на самых частых запросах были минимальные параметры (на глазок👀), конечно для этого бы придумать какой-то тул, но это уже другая история.
.
#clickhouse #2 orderby

Clickhouse-notes ⚡️
.
Продолжаем грузить клик
.
 В предыдущих notes рассмотрели загрузку:
- из файла CSV
- из датафрейма
- или сериализованых данных через HTTP интерфейс
.
Но есть и другие варианты: не всегда же источником являются файлы на локальной машине:
- S3 (наиболее частый случай, поставщик данных в S3 нас не интересует)
- Kafka и всё стримоподобное
- другие базы, например,  Postgres\MySQL
,
В каждом из кейсов могут свои оптимальные шаблоны, мы рассмотрим только возможности, имеющиеся в Clickhouse.
.
1️⃣Для загрузки из S3 используется S3 Engine (у нас еще будет задачка с S3 😉).
Тут всё довольно просто, нужны креды (access_key, access_secret_key), сам файл на S3 и табличная фукнция:

SELECT , toDate(now(), 'UTC') as row_inserion_date 
FROM s3('https://storage.yandexcloud.net/public-bucket-6/reddit/.parquet',  
'$AWS_ACCESS_KEY',  
'$AWS_SECRET_ACCESS_KEY',  
'Parquet',  
'id String, 
 noscript String, 
  score Nullable(Int64),  
  num_comments Nullable(Int64),  
  author String,  
  created_utc DateTime64,  
  url String,  
  upvote_ratio Float,  
  over_18 Bool,  
  edited Bool,  
  spoiler Bool,  
  stickied Bool,  
  subreddit_name_prefixed String  
  ') 
  SETTINGS input_format_allow_errors_num = 2000, 
  max_threads = 16; 

Что здесь хитрого:
- бакет public-bucket-6
- папка reddit
- дальше магические звезды ⭐️ (и это не распаковка Python)
- формат файла Parquet
Особенность при загрузке: необходимо указать схему данных (наименование поля и его тип, если схема не совпадает с тем, что есть на самом деле будет ошибка ).
⭐️() - нужны для pattern matching - чтобы загрузить данные из всех файлов в папке reddit - первая * = даты (партиционирование, обычно по датам), вторая * = просто все файлы с расширением .parquet.
.
В процесcе ETL обычно создаётся VIEW над S3 из нее данные загружаются в STG table (чтобы изменения на S3 не влияли на дальнейший процесс прогрузки).
ps: креды на чтение оставлю в комментах, чтобы вы могли попробовать😉.

Настройка, max_threads распараллеливает чтение😎
,
2️⃣ Следующий способ: движки баз данных или движки таблиц, на примере Postgres.
Довольно интересный способ, но в рабочей среде доводилось только экспериментировать. Например, при помощи движка таблицы можно легко получить структуру таблицы и перенести её в Clickhouse, так как он считает нужным:

CREATE TABLE stage.TMP_PG AS  SELECT * FROM postgresql('192.168.55.121:5432', 'tmp', 'kp_search_results', 'shpz', '12345', 'stage') WHERE 1; 

А вообще в команде принятым способом доставки данных из реляционок является Debezium (это уже история про потоковые данные). Но если нет инфры под Debezium и Кафка - движки баз данных и таблиц - хороший выбор👌
.
3️⃣ Последний кейс: потоковые данные из Кафка.
Хотя и в Clickhouse есть специальный движок, но даже экспериментировать не приходилось, тк для этого в команде есть Kafka-Connect.
ETL выглядит так: Kafka -> Kafka-Connect -> raw table in Clickhouse -> MV -> data table:
1. Kafka-Connect формирует собщение простой структуры:

record_timestamp DateTime64 - время записи сообщения
record_value JSON - тело сообщения

2. Структуру выше хранит raw table
3. MV парсит JSON сообщение и записывает их в data table
4. Пользователь ходит уже в data table

Вообще парадигма MV (materialized view)\Projection очень крутая, возможно надо ей уделить пару заметок.

#clickhouse #3 #extract #load

Маленький, но крайне полезный
.
Сегодня для вас маленький хук, при создании таблицы в клике по образу и подобию уже существующей не обязательно переписывать схему, достаточно указать эту аналогичную таблицу:

CREATE TABLE sandbox.table as stg.another_table
ENGINE=MergeTree
ORDER BY .....
SETTINGS ....

Это полезно, например при тестировании различных настроек: кодеки сжатия, партицирование, движки, секция ORDER BY.

#clickhouse #4 #liketable

С Новым годом, все причастные к миру данных🎄
.
ps: не будет долгих речей, просто будьте счастливы🦄

И напишите в комментах ваш любимый новогодний фильм 🎥
.
#newyear #2023

Праздники пришли когда не ждали
.
Раскопал интересный нюанс, в процессе парсинга дат. Попробуйте выполнить такой код:

SELECT 
parseDateTimeBestEffort('1950-01-01'); 

.
Результатом будите удивлены😳

#clickhouse #причуды

​​⚡️Clickhouse-notes - ответы (просвещаемся на каникулах 🙃)
.
Под одним из постом был комментарий:

Если не ошибаюсь, то стоит помнить, что order by сортирует только в рамках партиции, а не по всей таблице

.
Вероятно, предполагается парт данных, партиции - это про деление данных на части по какому-либо признаку (чаще всего встречается по времени, например, партицией может быть день\неделя\месяц).
Парт данных в контексте семейства Merge Tree - некоторый кусок неслитых данных, например, при вставке - это блок вставляемых данных (условно 1000 или 100000 строк).
.
Проверим применяется ли секция ORDER BY ко всей таблице.
Для экспериментов написал небольшой pipeline (github repo) для получения текущей погоды по API openweathermap.
DDL raw таблицы:

CREATE DATABASE cdc;

CREATE TABLE cdc.openweathermap_raw
(
record_timestamp DateTime64(6),
record_value String
)
ENGINE=MergeTree
ORDER BY (record_timestamp)
;

Pipeline выполняется по cron каждые 5 минут. Секция ORDER BY указывает сортировку по record_timestamp - это говорит, что сам парт будет отсортирован по указанному полю и все слитые парты (то есть таблица) также будут отсортированы по record_timestamp, что подтверждается экспериментально (см скрин 1 в комментах).
.
Какой нюанс стоит учитывать:
только что вставленный парт может не быть отсортирован (см скрин 2): только что вставленная строка оказывается в начале данных, спустя некоторое время она в фоне будет отсортирована.
.
GIthub репу еще будем использовать для разбора других интересных особенностей clickhouse.
#clickhouse #orderby

🦆Утки или 🐍 змеи ?
.
Мир данных изменяется очень быстро, не успел приготовить утку по-duckdb-овски, на повестке дня уже новый схожий инструмент и в твоём любимом питончике - chdb (от создателей Clickhouse).
.
Инструменты уже успели сравнить - тут 🤔
.
Парочку статей для ознакомления:
🦆 Переводная статья о duckdb
🐍 О внутреннем устройстве
🐍 Про chdb от Антона Жиянова (человек, который развивает и много разрабатывает для SQLite)
.
А вы что-то трогали, использовали?
ps: будет серия заметок про duckdb (построим небольшой пайплайн)
#duckdb #chdb