Apache Sqoop

Ознакомился Apache Sqoop. Это-проект который сейчас не поддерживается но очень хороший инструмент, где до сих пор используют. Основная задача этого инструмента-это import из РСУБД в Hadoop. Достаточно указать параметры подключение (host, user, password, db) и код sql. Также Sqoop использует инфраструктуру MapReduce для импорта и экспорта данных, что обеспечивает параллельный механизм и отказоустойчивость.

В текущем проекте, где я работаю Dataflow работает поверх Apache Sqoop и помогает импортить as is в Hadoop.

Статьи на Хабре
1. https://habr.com/ru/companies/neoflex/articles/646741/
2. https://habr.com/ru/companies/beeline/articles/679876/
3. https://coderlessons.com/tutorials/bolshie-dannye-i-analitika/uznaite-sqoop/sqoop-kratkoe-rukovodstvo

Пример импорта на sqoop

sqoop import \
--num-mappers 8 \
--fetch-size 100000 \
-Dmapreduce.task.timeout=0 \
-D mapreduce.map.memory.mb=16384 \
-Doraoop.import.partitions='"SERVICE_FEATURE_MAX"' \
--hcatalog-database customer360_stg \
--hcatalog-table customer_oracle_sqoop \
--connect jdbc:oracle:thin:@//10.31.178.66:1521/ORCL \
--username user \
--password password \
--table DWH.SERVICE_FEATURE \
--drop-and-create-hcatalog-table \
--hcatalog-storage-stanza 'stored as orcfile' \
--direct

NTILE

Ntile - это функция в sql. Она группирует строки на заданное количество частей.

Допустим поделим наши строки на три части в соответствии с уровнем зарплаты. Тогда SQL-запрос будет выглядеть так:

SELECT *,
NTILE(3) OVER(ORDER BY salary DESC) as backet_id
FROM work

—pyspark—-
w = Window.orderBy(F.desc("salary"))
df.withColumn("bucket_id", F.ntile(3).over(w)).show()

percentile_approx

percentile_approx - Агрегатная функция возвращает выражение, которое является наименьшим значением в упорядоченной группе (отсортированной от наименьшего к наибольшему) таким образом, что не больше percentile значений expr меньше значения или равно ему. Если percentile является массивом percentile_approx, возвращает приблизительный массив expr процентилей в указанном процентиле.

Параметр accuracy управляет точностью аппроксимации за счет памяти. Более высокое значение точности дает лучшую точность. 1.0/accuracy Это относительная погрешность аппроксимации.

Если DISTINCT задано значение , функция работает только с уникальным набором значений expr .

percentile_approx ( [ALL | DISTINCT ] expr, percentile [, accuracy] ) [FILTER ( WHERE cond ) ]

spark.sql(" SELECT percentile_approx(col, 0.75, 100) FROM VALUES (1), (2), (3), (4), (5), (6), (7) AS tab(col)").show()

Вопросы на собесе для DE.

У нас есть задача построить витрину данных об интересах абонентов. Интересы абонента разбиваются на 10 категорий. В качестве источников для определения интересов у нас есть следующие:

Информация о посещенных сайтах, о длительности сессии, объеме трафика
Информация о датах и количестве входящих/исходящих смс и их отправителях
Информация об исходящих/входящих звонках.
Разметка сайтов, отправителей смс, номеров телефонов на категории интересов поступает от бизнес-заказчиков в виде excel-файлов по почте, регламент обновления отсутствует.

Необходимо предложить архитектуру, наполнение витрины, а также организовать etl-процесс,
позволяющий как можно быстрее учитывать изменения в источниках данных и новую информацию от бизнес-заказчиков

Узнал интересное про фильтрацию по sysdate в Oracle.

Now  SYSDATE
Tomorow/ next day  SYSDATE + 1
Seven days from now  SYSDATE + 7
One hour from now  SYSDATE + 1/24
Three hours from now  SYSDATE + 3/24
An half hour from now  SYSDATE + 1/48
10 minutes from now  SYSDATE + 10/1440
30 seconds from now  SYSDATE + 30/86400
Tomorrow at 12 midnight  TRUNC(SYSDATE + 1)
Tomorrow at 8 AM  TRUNC(SYSDATE + 1) + 8/24
Next Monday at 12:00 noon  NEXT_DAY(TRUNC(SYSDATE), ‘MONDAY’) + 12/24
First day of the month at 12 midnight  TRUNC(LAST_DAY(SYSDATE ) + 1)
The next Monday, Wednesday or Friday at 9 a.m  TRUNC(LEAST(NEXT_DAY(sysdate,”MONDAY’ ‘ ),NEXT_DAY(sysdate,”WEDNESDAY”), NEXT_DAY(sysdate,”FRIDAY” ))) + (9/24)

Следующий пример как добавить час, минуту, секунду и день в sysdate.

select sysdate, sysdate+1/24, sysdate +1/1440, sysdate + 1/86400, sysdate+1 from dual;

Недавно в одном посте увидел, где разработчик указывал свои отрицательные стороны. Одна сторона совпало и со мной. Хотел бы поделиться и также поработать над этим.

Это не перерабатывать

Если приходится работать по вечерам или по выходным - значит что-то пошло не так.

Идея перерабатывать, чтобы получить повышение, а потом восстановить режим - такая же, как и посидеть немного на диете, сбросить пару килограмм, а потом вернуться к сладкому.

Из правила есть исключения - дежурства и дедлайны. Но задача - научиться более эффективно работать в рабочее время без вреда личному времени.

Хорошая статья про AQE (Adaptive Query Execution) в Spark

https://www.databricks.com/blog/2020/05/29/adaptive-query-execution-speeding-up-spark-sql-at-runtime.html

Также хорошее видео по оптимизации MERGE внутри спарка. https://www.youtube.com/watch?v=AIZjy6_K0ws

Всем привет.

Начинаю снова вести свой канал. Закончился один из сложных периодов в моей карьере.

Сейчас хотел оставить план запроса в спарк. Да он не поместился в один скрин. Иногда приходится такие запросы писать.

Работая в телекоммуникационной компании, я часто сталкиваюсь с новыми терминами и аббревиатурами. Хочу поделиться несколькими из них здесь.

1. IMSI - International Mobile Subscriber Number
2. MSISDN - (Mobile Station ISDN number) – номер абонента сети сотовой связи
3. IMEI - International Mobile Equipment Identity — международный идентификатор мобильного оборудования
4. CAMEL - это набор стандартов, действующих в сетях GSM, в соответствии с которыми операторы могут создавать новые услуги, не определенные в стандарте GSM.
5. USSD - Unstructured Supplementary Service Data
6. Эрланг (обозначение Эрл) - безразмерная единица интенсивности нагрузки (чаще всего телефонной нагрузки) или единица нагрузки, используемая для выражения величины нагрузки, требуемой для поддержания занятости одного устройства в течение определённого периода времени.
1 эрланг (1 Эрл) — соответствует непрерывному использованию одного голосового канала в течение 1 часа. То есть если абонент проговорил с другим абонентом в течение одного часа, то на телекоммуникационном оборудовании была создана нагрузка в один эрланг. Однако, если две пары абонентов последовательно проговорили друг с другом каждые в течение одного часа, то нагрузка на оборудовании была равна также один эрланг.

Записал свой первый видео ролик про свою специальность. Только на узбекском языке.

https://www.youtube.com/watch?v=GQsSShR9R0s

Сегодня погрузился как спарк управляет памятью executor’ов.

Напоминаю, что spark написан на Scala, поэтому все операции выполняются внутри JVM. Существует глобально две области памяти

- On-Heap Memory (spark.executor.memory)
- Off-Heap Memory + Overhead (spark.memory.offHeapsize + spark.executor.memoryOverhead) память вне кучи (Off-Heap Memory) – сегмент за пределами JVM, который иногда используется виртуальной машиной Java. Память вне кучи также может использоваться для хранения сериализованных датафреймов и RDD.

Основные вычисления и хранения данных происходит в On-Heap Memory. В свою очередь On-Heap Memory разделяется на следующие части:

1. Execution memory (=spark.execution.memory * spark.memory.fraction - spark.memory.storageFraction) - здесь происходит JOIN, Shuffle, Sorting, GroupBy
2. Storage memory (spark.memory.storageFraction=0.5) - хранятся закешированые данные и Broadcast переменные
3. User memory (=spark.executor.memory - spark.memoryFraction) - переменные определенные со стороны пользователя
4. Reserved memory (300Mb) - 300Mb зарезервированных данных

Пример:

`Spark.executor.memory=10Gb
spark.memory.fraction=0.8
spark.memory.storageFraction=0.5`

Execution memory => spark.executor.memory * spark.memory.fraction - <Storage Memory> => 10Gb * 0.8 - 4Gb = 4Gb
Storage memory => spark.executor.memory * spark.memory.fraction * spark.memory.storageFraction => 10Gb * 0.8 * 0.5 = 4Gb
User memory => spark.executor.memory - spark.memoryFraction => 10Gb - 300Mb - 10Gb*0.8 = 1.7Gb
Reserved memory => 300Mb

* Для примера 1Gb взял как 1000Мб, знаю что это 1024Мб

Также этот рисунок дает графическое описание, чтобы легче было воспринять

Решил интересную задачу от Яндекса. Хотел бы поделиться.

Дана таблица с логами пользователя в формате:

create table logs ( user_id text, time int );

INSERT INTO logs (user_id, time) values
('u1',0),
('u1',20),
('u2',30),
('u2',40),
('u1',40),
('u1',60),
('u2',65),
('u1',80),
('u1',130),
('u1',150),
('u1',200);

где user_id — id пользователя, time — время.

Активной сессией будем считать отрезок времени, в котором промежуток между действиями пользователя меньше или равен 30.

Нужно написать запрос, который найдет количество всех активных сессий пользователей, результат отсортировав по ключу user_id.

Вот самое решение:

WITH t AS
 (SELECT user_id,
         TIME,
         lag(TIME) over(PARTITION BY user_id ORDER BY TIME) AS prev_time
    FROM logs)
SELECT user_id, COUNT(*)
  FROM t
 WHERE prev_time IS NULL
    OR TIME - prev_time > 30
 GROUP BY user_id
 ORDER BY user_id

🎄 Advent of SQL

Начал решать задачи из Advent of SQL — это такой SQL-челлендж с интересными заданиями на каждый день. Решения выкладываю в репозиторий на GitHub:
🔗 Смотреть на GitHub

Если тоже участвуете или просто интересно посмотреть, заходите. Буду рад обратной связи или обсуждению! 😊