from pyspark import SparkContext
from pyspark.streaming import StreamingContext

# SparkContext и StreamingContext с интервалом в 5 секунд
sc = SparkContext("local[2]", "WordCountStreaming")
ssc = StreamingContext(sc, 5)

# gодключаемся к источнику (здесь TCP сокет)
lines = ssc.socketTextStream("localhost", 9999)
words = lines.flatMap(lambda line: line.split(" ")) -- конечно тут должен быть токенайзер, а не простой сплит.
word_counts = words.map(lambda word: (word, 1)).reduceByKey(lambda x, y: x + y) -- кстати отличный пример использование reduceByKey.
word_counts.pprint()  # выводим результат в консоль ну или конечно вставляем в таблицу но для простоты примера просто.

import logging

# уровень логирования
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("StreamingApp")

try:
    ssc.start()
    logger.info("Streaming started...")
    ssc.awaitTermination()
except Exception as e:
    logger.error("An error occurred in streaming: %s", e)
finally:
    ssc.stop(stopSparkContext=True, stopGraceFully=True)
    logger.info("Streaming stopped.")

optimized_word_counts = word_counts.coalesce(1)
optimized_word_counts.pprint()

import logging
from pyspark.sql import SparkSession
from pyspark.sql.functions import explode, split

# SparkSession для Structured Streaming
spark = SparkSession.builder.appName("StructuredWordCount").getOrCreate()

# подключаемся к Kafka
df = spark.readStream.format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "word_topic") \
    .load()

# конвертируем байты Kafka в строки - да это всегда дополнительные упражнения 🙂
df = df.selectExpr("CAST(value AS STRING)")

words = df.select(explode(split(df.value, " ")).alias("word")) # ну вы помните про токенайзер 
word_counts = words.groupBy("word").count()

# консольный вывод (можно вывести в HDFS, Cassandra и другие модные слова)
query = word_counts.writeStream \
    .outputMode("complete") \
    .format("console") \
    .option("checkpointLocation", "/path/to/checkpoint/dir") \
    .start()

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("StructuredStreamingApp")

try:
    logger.info("Starting Structured Streaming...")
    query.awaitTermination()
except Exception as e:
    logger.error("An error occurred: %s", e)
finally:
    spark.stop()
    logger.info("Structured Streaming stopped.")

# watermark 2 минуты если позже проигнорируем
word_counts = words.withWatermark("timestamp", "2 minutes") \
    .groupBy("word", window("timestamp", "10 minutes")) \
    .count()


query = word_counts.writeStream \
    .outputMode("complete") \
    .format("console") \
    .trigger(processingTime="10 seconds") \ # ставим не реал-тайм а раз в 10 секунд чтобы не насиловать систему все время а лишь раз в 10 секунд
    .option("checkpointLocation", "/path/to/checkpoint/dir") \
    .start()

## Сначала берем спарк и данные 
from pyspark.sql import SparkSession
from pyspark.ml.feature import VectorAssembler
from pyspark.ml.regression import LinearRegression

## Spark
spark = SparkSession.builder.appName("MLlibExample").getOrCreate()

data = spark.createDataFrame([
    (50, 2, 300),
    (60, 3, 400),
    (80, 3, 500),
    (100, 4, 700),
], ["square", "rooms", "price"])

## асемблируем -- если не знаете что это такое не страшно.
assembler = VectorAssembler(inputCols=["square", "rooms"], outputCol="features")
data = assembler.transform(data)

## Учим нашу простую модель

# Инициализация и обучение модели
lr = LinearRegression(featuresCol="features", labelCol="price")
model = lr.fit(data)

## Предсказываем цену

predictions = model.transform(data)
predictions.select("features", "price", "prediction").show()

Ответ почему оно не работет всегда в зеркале в 99.9% случаев.

import threading -- для контроля потоков
import time -- для времени ведь его всегда так мало

# Тут будем хранить какое последнее число посчитали простым или четным
progress = {"prime": None, "even": None}

stop_prime = threading.Event()

def prime_worker():
    n = 2
    while not stop_prime.is_set():
        is_prime = all(n % i != 0 for i in range(2, int(n**0.5) + 1))
        if is_prime:
            progress["prime"] = n
        n += 1
        time.sleep(0.01)

prime_thread = threading.Thread(target=prime_worker, daemon=True)
prime_thread.start()

print("Поток prime_worker запущен")

stop_even = threading.Event()

def even_worker():
    n = 0
    while not stop_even.is_set():
        if n % 2 == 0:
            progress["even"] = n
        n += 1
        time.sleep(0.01)

even_thread = threading.Thread(target=even_worker, daemon=True)
even_thread.start()

print("Поток even_worker запущен")

progress

cd <имя папки> #чтобы перейти в папку в какую то.

cd .. #чтобы выйти на уровень выше из папки.

ls -la #это просмотр содержимого в папке в которой вы находитесь, выпадает список файлов.

git switch -c <ИМЯ ВЕТКИ> -- это создает новую ветку с именем что вы придумали.

git add <ИМЯ ФАЙЛА ИЛИ ПАПКИ> -- это добавляет в вашу ветки файлы что вы изменили, можно через пробел добавить несколько.

git commit -m 'Например: Добавил конфиг' -- Это сам коммит и коментарий в котором вы пишите что вы там добавили, чтобы ваши коллеги понимали что вы там сделали 🙂

## Но ваши файлы еще не в репозитории это только коммит изменений. Теперь делаем пуш в репозиторий.

git push --set-upstream origin <ИМЯ ВЕТКИ> -- имя ветки что вы указывали в начале.

default_timezone = utc

executor = LocalExecutor

parallelism = 16

max_active_tasks_per_dag = 4

max_active_runs_per_dag = 8

payload = {"user": {"name": "Alice"}}

# Явное обращение по ключу
nickname = payload["user"]["nickname"]          # KeyError, DAG падает

# Спокойный вариант c .get()
nickname = payload.get("user", {}).get("nickname", "Anon")

# "Если ты сигма" - способ (когда поле может отсутствовать целиком)
nickname = (payload.get("user") or {}).get("nickname", "Anon")

if "key" in resp:
    value = resp["key"]

# одна строчка
value = resp.get("key")

-- Сколько авторов имеют > 3 статей
SELECT   author_id,
         COUNT(*) AS article_cnt
FROM     articles
GROUP BY author_id
HAVING   COUNT(*) > 3;

-- Хотим все ордера + успешные доставки
SELECT  o.id,
        d.tracking_code
FROM    orders o
LEFT JOIN deliveries d
  ON    d.order_id = o.id
  AND   d.status = 'shipped'   -- фильтр СРАЗУ при соединении
WHERE   o.created_at >= current_date - interval '30 days';

WITH active AS (
  SELECT *
  FROM   sessions
  WHERE  started_at >= current_date - 7
)
SELECT  user_id,
        started_at,
        ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY started_at) AS rn
FROM    active
ORDER BY user_id, rn;

-- Ищем заказы c дорогими товарами и уже оплаченным счётом
SELECT  o.id,
        o.created_at,
        c.total_amount,
        c.status    AS charge_status,
        p.name      AS product_name,
        p.price
FROM    orders      o
JOIN    order_items oi  ON oi.order_id   = o.id
JOIN    products    p   ON p.id          = oi.product_id
JOIN    charges     c   ON c.order_id    = o.id
WHERE   p.price      > 1000                -- фильтр по цене
  AND   c.status     = 'paid'              -- фильтр по чеку
  AND   o.created_at >= current_date - interval '90 days';