Мы потеряли миллионы записей: вот как я его поймал, исправил и построил систему мониторинга в одном
data data-science monitoring-system data-platform pandas python-script snowflake-via-python monitoring-table building-a-monitoring-system

Мы потеряли миллионы записей: вот как я его поймал, исправил и построил систему мониторинга в одном

11 июня 2025 г.

Когда вы инженер по поддержке, аналитик и пожарный в одном человеке.

Контекст: большие данные, реальные клиенты, реальные проблемы

Я работаю в команде Data Platform, поддерживающей клиентов. Наш стек включает в себяСнежинкаВBigQuery,и продукт разоблачаетмиллионы точек данных ежедневно: Метрики клиентов, аналитика на уровне магазина, исторические тенденции и т. Д.

Итак, когда кто -то сказал,«Почему моя панель пуста?»Это была не просто ошибка пользовательского интерфейса. Что -то было сломано глубоко в трубопроводе данных.

Спойлер:Целые куски данных никогда не попадали в нашу платформуПолем

Шаг 1: Исследование черной дыры

Первая подсказка пришла от команды доставки продукта: в некоторых учетных записях были неполные наборы данных. Мы быстро проследили поток:

  • Необработанные данные введеныMysql
  • Он должен был перейти кGoogle Cloud Storage (GCS)
  • Но не попал вСнежинка

Нет тревоги. Нет оповещений. Просто ... безмолв.

Сигнал в передаче GCS → Snowflake молча пропускал записи из -за неэлементной схемы и отсутствия проверки.

Шаг 2: Создайте систему мониторинга (через неделю)

Я решил построить легкий, но мощныйМониторинг трубопровода с использованием Python и SnowflakeПолем Не ожидание долгосрочного исправления продукта, потому что это должно было работатьсейчасПолем

Шаг 2.1: Подключитесь к снежинкам через Python

#a connector
import snowflake.connector
import pandas as pd

conn = snowflake.connector.connect(
    user='user',
    password='password',
    account='account_id',
    warehouse='warehouse',
    database='db',
    schema='schema'
)
#just an example
query = """
SELECT customer_id, module, max(event_date) as last_date
FROM analytics_table
GROUP BY customer_id, module
"""
df = pd.read_sql(query, conn)

Шаг 2.2: Обнаружение задержек с помощью панд

df["lag_days"] = (pd.Timestamp.today() - pd.to_datetime(df["last_date"])).dt.days
df["status"] = df["lag_days"].apply(lambda x: "valid" if x <= 1 else "delayed")

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

Шаг 2.3: храните результаты в новой таблице мониторинга

from sqlalchemy import create_engine
#example
engine = create_engine(
    'snowflake://user:password@account/db/schema?warehouse=warehouse'
)

df.to_sql("monitoring_status", con=engine, if_exists="replace", index=False)

Шаг 3: Покажите боль визуально

С использованиемLooker StudioЯ подключился прямо к таблице мониторинга.

Я построил приборную панель, которая показала:

  • % модулей, задержанных на одного клиента
  • Время с момента последнего успешного синхронизации
  • Новые проблемы, когда они появились

Это сделало лидерствовидетьРазрыв и распределяет ресурсы.

Шаг 4: Исправьте основную причину

Тем временем мы установили сломанный анализатор и перезапустили все затронутые трубопроводы. Мы также задним числом обработали все потерянные данные. Благодаря мониторингу, он подтвердил свой полный.

Результаты

  • Восстановлен100% потерянных данных
  • Создал аВ реальном времени, всегда мониторинг мониторинга
  • Снижение будущих слепых пятен за счет автоматической проверки
  • ПредотвращеноКлиентские эскалацииеще до того, как они заметили

Последние мысли

Иногда все, что вам нужно, - это интеллектуальный инженер поддержки, сценарий Python и четкое представление о том, что не так.

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

Оригинал

Recent Post

Categories

PREVIOUS ARTICLE
NEXT ARTICLE