Список действий и трансформаций

Решил временно сместить тематику и рассказать о каких то простых вещах:

Например иногда в шутку на собесах или в неформальной обстановке Python разработчики рассказывают/хвастаются возможность написать калькулятор в одну строчку. (На Java без библиотек так естественно не получится)

Итак, таким образом вы можете создать калькулятор в одну строку (правда для постоянного использования его нужно будет перезапускать, но можно написать еще строчку и все будет работать вечно)

print(eval(input()))

На собеседовании могут спросить о способах удаления элементов из Python листа, приведу примеры основных и экзотических способов. Когда у меня спросили я назвал только четыре.

используя ключевое слово del. (Нужно так же знать индекс элемента)

list1 = ["a", "b", "c", "d"]
del list1[1]

Используя метод remove (удалит первый встретившийся в листе элемент)

list2 = ["a", "b", "c", "d"]
list2.remove("b")

Используя метод pop (нужно указать индекс, иначе удалит последний элемент)

list3 = ["a", "b", "c", "d"]
list3.pop(1)

Используя синтаксический сахар - list comprehension:

list4 = ["a", "b", "c", "d"]
list4 = [i for i in list4 if i != "b"]

Используя лямбда-функцию (так же известную как анонимную) в функции filter

list5 = ["a", "b", "c", "d"]
list5 = list(filter(lambda i: i != "b", list5))

Встроенными инструментами работы с листами в Python

list6 = ["a", "b", "c", "d"]
list6[1:2] = []

Важно помнить что все эти способы кроме лямбда функции и list comprehension - модифицируют исходный лист, лямбда функция и list comprehension создают новый лист.

А теперь к совсем экзотическим, но в зависимости от постановки задачи их тоже можно упомянуть.

метод clear - полностью очистит ваш лист, что удалит как нужный элемент так и все остальные.🤡

list7 = ["a", "b", "c", "d"]
list7.clear()

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

first_list = ["a", "b", "c", "d", "c"]
set_list = set(first_list)
set_list.discard("b")
main_list = list(set_list)
print(main_list)

Как производить расчет прошлых дат при создании DAG в Apache Airflow ? 🤔

В Apache Airflow существует такое понятие как backfilling - планирование расчета дат в прошлом. Очень распространенный use case для тех кто работает с данными.

Есть несколько способов триггернуть его
1) Через консоль (не всегда доступен тк вам могут и не дать прямого доступа на хост чтобы вы ничего не положили, тем не менее использовать все равно можно, расскажу о двух хитростях в следующих постах)

airflow dags backfill \
    —start-date START_DATE \
    —end-date END_DATE \
    dag_id

2) через настройки дага
☝️ важно помнить - зачастую так же необходимо указать параметр отвечающий за необходимость запуска/не запуска дополнительных потоков

dag = DAG(
          your_job_id, 
          default_args=your_dag_args, 
          schedule_interval=your_scheduler, 
          max_active_runs=1,
          catchup=True
      )

так с помощью параметра catchup (активного по умолчанию, вы можете контролировать этот самый бэкфиллинг)

параметр max_active_runs, как очевидно из названия контролирует число потоков отсчитывающих прошлые дни. Если поставить 1, то будете постепенно отсчитывать по 1 дню, что хорошо, понятно и предсказуемо для вашего кода.
В противном случае, в зависимости и от логики вашей таблицы вы рискуете все испортить)
Например если у вас есть staging таблички какие то - то вы не можете сразу несколько дней вместе рассчитывать, без дополнительной логики.

В общем этот backfilling считается одной из сильнейших сторон Apache airflow. Это спрашивают на собесах и это мощный инструмент, что вам наверняка пригодится если вы используете Airflow.

🧠 топ консольных команд что я использую в работе:

Этой командой вы можете добавить к своей спарк сессии джарник с какой либо библиотекой и использовать эту библиотеку в своем спарк контексте

%AddJar file:/home/my_jar.jar

Это команды для очистки технических папочек что не нужны вам и заполняют память кластера. Они очищаются и сами по себе спустя время, однако при особо тяжелом расчете и падении - вы можете в моменте очень сильно засорить память кластера и себе и коллегам. Не беспокойтесь за вами наверняка придут люди сопровождающие кластер и спросят почему в вашей папке с мусором лежит 1 Тб данных

hdfs dfs -rm -r -skipTrash .sparkStaging

hdfs dfs -rm -r -skipTrash .Trash

Просто положить файлик на HDFS чтобы потом считать

hdfs dfs -put file.txt

убить подвисшее приложение, или приложение что ест особо много ресурсов

yarn app -kill application_1532419910561_0001

узнать кто убил ваше приложение

yarn app -status application_1532419910561_0001 | grep killed

в одну команду обновить ваш тикет в керберосе (если просто кинит написать то потом нужно будет отдельно пароль вводить)

echo MyPassword | kinit -p myUser

положить какую либо папочку с вашего локального компьютера на какой либо хост (на том хосте должен лежать ваш ключик)

scp -r local-directory username@remotehost.ru:remote_directory

Интересный пост моего коллеги об использовании ИИ в разработке.
Стоит отметить что это очень удобно для разработчиков из России, поскольку от OpenAI на Россию наложена анафема (надеемся что не навсегда) - и таким образом можно получить неплохой инструмент без шарад с использование VPN, левых сим-карт, продавцов из не подсанкционных стран итп.
По моим ощущениям это все же немного слабее заграничных инструментов вроде ChatGPT и GitHub CoPilot, но кто знает что будет дальше.
В нынешней итерации это уже итак неплохой инструмент.

⚡️⚡️⚡️Хочу обратить ваше внимание на канал моего знакомого - Евгения, дата инженера из одной крупной российской компании.
Он очень харизматичный и активный парень, записывает интервью с представителями разных специальностей и рангов в мире IT.

Вот его интервью с Middle Java Developer.

https://youtu.be/Kv8r4Y9dpWM?si=NXrvhcfbwD8neArR

Так же он активно посещает различные IT мероприятия в Москве и рассказывает о них.
При всем при этом успевая продуктивно работать и очень доходчиво объяснять сложные темы в своих постах. Я часто узнаю в этих постах что-то новое и ликвидирую пробелы в собственных знаниях.

https://news.1rj.ru/str/halltape_data

🐘 Сейчас в рамках решения рабочих задач пользуюсь одним относительно старым планировщиком рабочих процессов.
Здесь он представлен как Apache Oozie.
Это приложение позволяет создавать достаточно сложные и разветвленные направленные ациклические графы с задачами.
В отличии от airflow вам не нужно знать Python.
Хотя нужно знать очень много чего другого.
По моему мнению у этого инструмента порог вхождения значительно выше чем у Airflow.
Вам нужно самим конфигурировать кучу файлов, вы должны четко понимать что вы делаете, знать компоненты экосистемы, ваш TaskTracker, NameNode, самому собрать Джарник со Спарк приложением, положить его на кластер, все настроить, установить на кластер, поднять сам хадуп кластер если его нет. И в общем это все тянется в кучу очень нетривиальных задач.

В то время как в airflow достаточно сделать pip install и можно уже в целом ехать.

Сам инструмент использует xml разметку. Может запускать из коробки spark приложения, так же есть свой аналог BashOperator в airflow.

Этот мастодонт никуда еще не денется и уметь им пользоваться важно, особенно в больших компаниях что уже вложили кучу денег в разработку и в эти инструменты.

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

Ну и забавный момент. Изначально это слово Уузи исходит из Мьянмы. Там это тренеры и погонщики слонов. Что забавно и создателями предполагалось как игра слов - ведь ваш хадуп кластер имеет логотип слона. А уузи это погонщик этого слона что заставляет его бегать и решать задачи.

Здравствуйте подписчики, выдался очень плотный спринт (это как правило двухнедельный цикл за который команда разработки должна выполнить ряд задач и предоставить бизнесу(заказчику) инкремент). Нужно было быстро усваивать кучу знаний в товарных количествах, что в целом хорошо.

В частности в прошлом и в текущем спринте работаю с планировщиком задач о котором говорил в прошлом посте.
Давайте остановимся на нем поподробнее.
Фактически oozie это Java приложение что запускает другие приложения и помогает выстроить сложные и разветвленные зависимости.

Что нужно для запуска?

Если речь идет о минимальнейших условиях, то вам нужно чтобы у вас было два файла

job.properties - файл с описанием переменных вашего хадуп кластера и путем до workflow.xml

oozie.use.system.libpath=True
oozie.wf.applijobcation.path=hdfs://mycluster/oozie-app/myapp/
nameNode=YOURNAMENODE:port
jobTracker=YOURYARN:8032

workflow.xml
Сам файл с описанием потока. Он не ограничивается одним действием, но лично я бы рекомендовал для соблюдения атомарности и облегчения дебаг не делать больше одного действия в рамках одного workflow.xml. (Наследие airflow хаха) Вы можете запускать все что угодно отсюда (вернее все что установлено на ваш кластер): башники, питон скрипты, джарники, Спарк приложения (основной юз кейс), пиг Джобы, хайв скрипты итд

<workflow-app name="Abobus" xmlns="uri:oozie:workflow:0.5">
    <start to="do-sh-run"/>
    <action name="do-sh-run">
        <shell xmlns="uri:oozie:shell-action:0.1">
            <job-tracker>${jobTracker}</job-tracker>
            <name-node>${nameNode}</name-node>
            <exec>/bin/bash</exec>
            <argument>test.sh</argument>
            <file>test.sh#test.sh</file>
              <capture-output/>
        </shell>
        <ok to="End"/>
        <error to="Kill"/>
    </action>
    <kill name="Kill">
        <message>Action failed, error message[${wf:errorMessage(wf:lastErrorNode())}]</message>
    </kill>
    <end name="End"/>
</workflow-app>

Далее в зависимости от версии oozie вы можете либо пойти в веб интерфейс мониторить свои джобы и читать логи (это можно делать напрямую не заходя в ярн если у вас было Спарк приложение), либо читать через консоль узи.

На бумаге все звучит просто, но фактически, тут множество подводных камней, особенно когда в дело вступают различные ролевые модели. Например у меня на днях сложилась ситуация что мой собственный юзер не мог писать в собственного хомяка (home/username - эта фраза является программистским жаргоном).
Поэтому при работе с такими сложными многокомпонетными приложениями и системами, нужно набраться воздуха, вдохнуть и выдохнуть - а иначе все, или станете невротиком или выгорите.

💢 недавно столкнулся с неприятной проблемой при работе с S3
Но для начала вкратце расскажу о том что же такое S3.
S3 (Simple Storage Service) три Эс если по-русски - это объектное хранилище (НЕ ФАЙЛОВАЯ СИСТЕМА)
и облачный сервис изначально разработанный Amazon Web Services (AWS).
Но фраза S3 уже стала нарицательной и я заметил что так называют хранилища даже не относящиеся к амазону и не использующие его сервис.
Строго говоря вся проблема этой путаницы в закрытом исходном коде, - умные американцы не хотят раскрывать свой код, что стабильно приносит очень хорошие деньги, но технология классная поэтому появляются клоны вроде Nimbus, Fake S3, HPE Helion Eucalyptus и Scality S3.
С одним из таких клонов я и работал, для простоты и я продолжу называть его S3.
Итак S3 позволяет пользователям загружать и отправлять данные ЛЮБОГО типа по вебу.
S3 очень ценят за его низкую (в рамках большого бизнеса) цену,
его используют для бэкапов, контент платформ, CDNов, сохранений данных high performance приложений, и конечно для Биг Даты.
Так же считается что под капотом Яндекс Диска и Гугл диска некие свои аналоги aws S3.

С какой же именно проблемой я столкнулся и почему она была неприятной:
В ПРОМ таблице перестали отображаться данные что загружались airflow дагом.
быстрое обращение к s3 с нужным префиксом показало что там этих файлов нет
отправитель файлов в s3 показал свои логи согласно которым файлы были точно успешно отправлены

спрашивается где файлы?

в общем я начал дебажить и обнаружил что при прямом обращении к файлам из логов они находятся

и встал вопрос где же произошла проблема.

в эйрфлоу даге все горит зеленым

контроллер отправляющий в s3 так же говорит что все было успешно отправлено

в обшем, после отправления самых разных вариантов запросов к s3 я все же смог обнаружить эти файлы так

import boto3


S3_ENDPOINT = 'fake-data'
S3_BUCKET = 'fake-data'
S3_REGION = 'fake-data'
S3_ACCESS_KEY_ID = 'fake-data'
S3_SECRET_ACCESS_KEY = 'fake-data'
PREFIX_TO_ACCESS = 'mybucket/fake-data/'

sesssion = boto3.Session(S3_ACCESS_KEY_ID, S3_SECRET_ACCESS_KEY)
s3 = sesssion.resource('s3', endpoint_url=S3_ENDPOINT)
bucketName = S3_BUCKET
bucket = s3.Bucket(bucketName)

list_of_lost_files = []

for obj in bucket.objects.filter(Prefix=PREFIX_TO_ACCESS):
    list_of_lost_files.append(obj.key)

print(list_of_lost_files)

почему же не исходный способ не показал эти файлы?

import boto3


headers = {'Content-Type': 'application/sql'}

S3_ENDPOINT = 'fake-data'
S3_BUCKET = 'fake-data'
S3_REGION = 'fake-data'
S3_ACCESS_KEY_ID = 'fake-data'
S3_SECRET_ACCESS_KEY = 'fake-data'
PREFIX_TO_ACCESS = 'mybucket/fake-data/'

session = requests.Session()

s3_client = boto3.client(
    service_name='s3',
    endpoint_url=S3_ENDPOINT,
    region_name=S3_REGION,
    aws_access_key_id=S3_ACCESS_KEY_ID,
    aws_secret_access_key=S3_SECRET_ACCESS_KEY,
)

response = s3_client.list_objects_v2(
    Bucket=S3_BUCKET,
    Prefix=PREFIX_TO_ACCESS)


for obj in response.get('Contents', []):
    print('\n')
    print(obj['Key'])
    print(obj['LastModified'])

избавлю вас от мучений - все дело в методе list_objects_v2 и в bucket.objects

всегда помогает чтение доки и помогло мне при решении этой проблемы
https://boto3.amazonaws.com/v1/documentation/api/latest/reference/services/s3/client/list_objects_v2.html
чтение доки подсказало что метод list_objects_v2() не может вернуть больше 1000 объектов как не старайся

MaxKeys (integer) – Sets the maximum number of keys returned in the response.
By default, the action returns up to 1,000 key names.
The response might contain fewer keys but will never contain more.

и с моим префиксом накопилось уже больше 1000 объектов
следовательно он возвращал только старые хотя новые уже были готовы и лежали рядом
Ситуацию осложняло то что мы не удаляем данные из s3 а только читаем

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

🦠 Услышал такую интересную фразу от коллеги из другого отдела

«Код вы пишите один раз - а читают его потом много раз, вы в том числе. Поэтому всегда лучше писать его сразу как следует»

Вам может показаться это душными понятиями каких то старых дедушек, но с течением времени я понял насколько это важно.

Вот мой личный топ АНТИ ПАТТЕРНОВ в разработке:

1) В самом начале я имел неосторожность использовать ключевые слова в названии переменных, думаю тут не стоит что то еще говорить - ваша программа будет просто вести себя нестабильно если вы будете использовать ключевые слова как переменные. в моем случае я использовал слово type в Python2

2) Писать огромные функции с большим числом аргументов. Их так же сложно читать и дебажить, в целом есть даже такое правило что в названии вашей функции не должно быть слов вроде and or with итд - то есть нужна строгая модульность: одна функция выполняет одну задачу
так же есть правило не больше пяти аргументов на функцию.

3) Не писать юнит тесты - тут все просто при добавлении нового функционала вам не придется как обезьянке заново все проверять вручную теряя время. Что гипер важно.

4) Не использовать стандарты языка (в Python этот стандарты PEP) они унифицируют то какие отступы нужно делать между импортами, как называть и не называть переменные, как форматировать строки и многое многое другое

5) применять очень большие try/except конструкции - при особых условиях из за них можно просто потерять ошибку в логах, а так же это крайне неудобно дебажить.

6) использовать непонятные переменные вроде i,j,a,b итд. Вам будет неудобно в будущем вникать что же они значат и например по чему же предполагается тут итерироваться.

Поговорим о так называемой БАЗЕ - *Медленно меняющиеся измерения (Slowly Changing Dimensions)* - ЭсКаДэ - это детище методолгии Кимбалла.
по сути это таблицы, в которых некоторые атрибуты могут изменить свои значения по истечении некоторого периода времени, причем частота таких изменений является небольшой.
Интересно это с точки зрения организации версионирования данных в вашей таблице.

SCD 0
данные после первого попадания в таблицу далее никогда не изменяются. Нужен лишь как нулевая точка отсчета для методологии SCD. По сути, вообще не SCD.

SCD 1
это обычная перезапись старых данных новыми. В чистом виде этот метод тоже не содержит версионности и используется лишь там, где история фактически не нужна

SCD 2
для каждой версии создается отдельная запись в таблице с добавлением поля-ключевого атрибута данной версии.

SCD 3
В самой записи содержатся дополнительные поля для предыдущих значений атрибута. При получении новых данных, старые данные перезаписываются текущими значениями.

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

SCD 5
Методика SCD 5 основана на мини-измерении SCD 4 путем внедрения ключа мини-измерения «текущий профиль» в базовое измерение, которое перезаписывается как атрибут SCD 1. Поэтому в общем то SCD 5 (4+1), это позволяет получить доступ к назначенным в данный момент значениям мини-измерения вместе с другими значениями базового измерения без привязки через таблицу фактов.

SCD 6
Комбинация 1, 2 и 3 методов и предназначен для ситуаций, которые они не учитывают или для большего удобства работы с данными.

SCD 7
это другой способ достижения того же, что и SCD 6, где вы поддерживаете версию значений SCD 1 отдельно от версии значений SCD 2. Часто версия SCD 1 создается с использованием представления SCD 2. Имея оба ключа в факте, вы можете узнать, как обстояли дела на момент факта, а также как все обстояло на основе текущих версий измерений. Это позволяет избежать необходимости обновлять старые значения в соответствии с текущим состоянием.

Расскажу об особенностях использования CRONa. 🕰️

Иногда бывает что нужно запускать какой то скрипт регулярно, а полноценные ETL фреймворки использовать невозможно/слишком затратно итп. 🌚

Итак что нужно сделать чтобы использовать утилиту CRON на Linux.

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

Сделать этот скрипт исполняемым. Например так

Chmod -x  my_noscript.sh
Chmod 777 my_file.py

Убедиться в том что пути(если они есть) в вашем скрипе абсолютные.

Что это значит?
Компьютер не знает о ваших локальных файловых системах и не будет запускаться из под папки юзеров где вы скорее всего отлаживали свой код - а значит нужно указать везде абсолютные пути если они есть.
Еще лучше это делать программмно. Используя стандартный модуль os или sys в случае Python.

Важный момент! ‼️
Если вы подменяете файл не забывать его так же делать исполняемым.
При подкладывании другого файлу и переименовании меняется номер инода - соответсвенно для компьютера это совершенно другой файл и вы можете столкнуться с проблемой при запуске тк прав 777 у него уже нет (777 - максимальный доступ к файлу для всех пользователей и групп пользователей)

Про то как именно настраивать время запуска CRON - можно почитать здесь.

https://crontab.guru/every-1-minute

Итак что же здесь происходит? 🤯

На самом деле тут вступают в дело несколько особенностей языка python.

Во первых
хоть на первый взгляд кажется, что функция print_list и была объявлена раньше чем был создан лист new_list - это не мешает коду быть корректно выполненным.
До непосредственного вызова функции print_list - python просто создает привязку имени функции. Никакой код не выполняется, кроме анализа синтаксиса.
Обратите внимание, что это не похоже на PHP или JavaScript, где функции 'поднимаются' -т.е. любые определения функций обрабатываются и анализируются раньше всего остального. В PHP и JavaScript совершенно законно определять функции в коде ниже, чем то место, где они вызываются.

Таким образом, синтаксический анализатор не дошел до выполнения print_list() пока его непосредственно не вызвали и соответственно не упал с NameError: name 'new_list' is not defined. Более того - если бы мы не вызвали в конце эту функцию print_list() и понаписали в нее еще больше несуществующих листов вроде этого😬

def print_list(some_list):
    for element in sudden_list:
        print(element)
    for element in mega_list:
        print(element)
    for element in super_list:
        print(element)


print('Hello')

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

во вторых - уже про область видимости. new_list находится в глобальной области видимости, то есть изнутри функции print_list мы можем к нему обращаться, поскольку из локальной области видимости видно то, что происходит в глобальной области видимости (но очевидно не наоборот).

😡😡😡 Антитоп Антипаттернов при написании кода в Питоне
Кое что вобрал от своих коллег, к чему то пришел сам:
итак

1) Переменные названы непонятно для стороннего человека или не до конца объясняют суть переменной

hrs = 5
weeks = 3
for p in products:
    print(p)

2) Неиспользуемые переменные остаются в коде

for i, product in enumerate(products): 
  print(i)

3) «Магические» числа в коде

for I in range(24):

4) Переменные названы неконсистентно

kids_card =''
det_union_month=''

5) Функция слишком длинная и принимает 5 и более аргументов

6) Функция выполняет несколько дел:

def create_and_push():

7) Используется булевый флаг для изменения поведения функции

def round_number(value: float, up: bool) -> float:

8) Такие артефакты в коде

def hello_name(name):
    print(name)

9) Не используем названия аргументов при вызове функции

tournament_search("Chess", False, 2023, True)

10) Проверяем на правильность, а не "пробуем" выполнить код с полученными данными

if not os.path.isfile(filename):
    print('File does not exist.')                                    
if not os.access(filename, os.W_OK):
    print('You dont have write permission.')
with open(filename, 'w') as file:
    file.write('Hello, World!')
    return(f"Hello {name}")

11) Не используем менеджер контекстов

file = open(file_path, 'w')
file.write('Hello, World!')
file.close()

12) Комментируем слишком подробно или не комментируем совсем