diff --git a/scouter/activities/pydruid.py b/scouter/activities/pydruid.py index d1aa0ee..473c173 100644 --- a/scouter/activities/pydruid.py +++ b/scouter/activities/pydruid.py @@ -6,6 +6,8 @@ with workflow.unsafe.imports_passed_through(): from datetime import datetime, timedelta from pydruid.client import PyDruid from pydruid.query import QueryBuilder + from sqlalchemy.engine import create_engine + from sqlalchemy import MetaData, Table, select, text from sientia_do.temporal.activities.base import BaseActivity from sientia_do.notifications.handlers import NotificationHandler from sientia_do.temporal.utils.logger import Logger @@ -13,15 +15,12 @@ with workflow.unsafe.imports_passed_through(): class Druid(BaseActivity): def __init__(self, host: str, port: int, - logger: Logger, notification_handler: NotificationHandler, - endpoint: str = "druid/v2"): + logger: Logger, notification_handler: NotificationHandler): self.host = host self.port = port - self.endpoint = endpoint - self.client = PyDruid( - f"http://{self.host}:{self.port}", {self.endpoint} - ) + self.engine = create_engine( + f'druid://{self.host}:{self.port}/druid/v2/sql/') logger.info( f"Druid client initialized with host: {self.host}, port: {self.port}") @@ -41,36 +40,25 @@ class Druid(BaseActivity): """ metadata = input_data['metadata'] datasource = f"raw_{input_data['schedule_name']}" - last_data_timestamp_str = input_data['last_data_timestamp'] - if last_data_timestamp_str is None: - last_data_timestamp = datetime(1970, 1, 1, 0, 0, 0) - else: - last_data_timestamp = datetime.strptime( - last_data_timestamp_str, "%Y-%m-%d %H:%M:%S.%f") - + last_data_timestamp = input_data['last_data_timestamp'] self.debug( f"Loading data from Druid: {input_data}", metadata=metadata) - end_time = datetime(9999, 12, 31, 23, 59, 59) - - interval = f"{last_data_timestamp.isoformat()}Z/{end_time.isoformat()}Z" + query = f'"__time" > TIMESTAMP \'{last_data_timestamp}\'' self.info( - f"Loading data from Druid: {datasource} with interval: {interval}" + f"Loading data from Druid: {datasource} with query: {query}" ) - builder = QueryBuilder() + places = Table(datasource, MetaData(), autoload_with=self.engine) + stmt = select(places).where(text(query)) - query = builder.scan( - { - "datasource": datasource, - "intervals": interval, - "columns": ["timestamp", "value", "tag"], - "limit": 10000, - } - ) + result = pd.read_sql(stmt, self.engine) - result = query.export_pandas() + result["inserted_at"] = pd.to_datetime(result["__time"]).dt.strftime( + "%Y-%m-%d %H:%M:%S.%f") + + result.drop(columns=["__time"], inplace=True) self.info( f"Loaded {len(result)} rows from Druid" diff --git a/test.ipynb b/test.ipynb new file mode 100644 index 0000000..fb78908 --- /dev/null +++ b/test.ipynb @@ -0,0 +1,267 @@ +{ + "cells": [ + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "from sqlalchemy.engine import create_engine\n", + "\n", + "engine = create_engine('druid://localhost:8082/druid/v2/sql/')" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "from sqlalchemy import MetaData, Table\n", + "\n", + "metadata = MetaData()\n", + "places = Table('raw_scouter-opcua-orchestrated-pipeline', metadata, autoload_with=engine)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "from sqlalchemy import select\n", + "\n", + "stmt = select(places)\n", + "with engine.connect() as conn:\n", + " result = conn.execute(stmt)\n", + " for row in result:\n", + " print(row)" + ] + }, + { + "cell_type": "code", + "execution_count": 5, + "metadata": {}, + "outputs": [ + { + "name": "stderr", + "output_type": "stream", + "text": [ + "/tmp/ipykernel_30057/1086103244.py:5: SADeprecationWarning: The dbapi() classmethod on dialect classes has been renamed to import_dbapi(). Implement an import_dbapi() classmethod directly on class to remove this warning; the old .dbapi() classmethod may be maintained for backwards compatibility.\n", + " engine = create_engine('druid://localhost:8082/druid/v2/sql/')\n", + "/home/grezewave/Documents/projects/sientia/sientia-dataops-scouter_temporal/venv/lib/python3.11/site-packages/pydruid/db/sqlalchemy.py:188: SAWarning: Dialect druid:rest will not make use of SQL compilation caching as it does not set the 'supports_statement_cache' attribute to ``True``. This can have significant performance implications including some performance degradations in comparison to prior SQLAlchemy versions. Dialect maintainers should seek to set this attribute to True after appropriate development and testing for SQLAlchemy 1.4 caching support. Alternatively, this attribute may be set to False which will disable this warning. (Background on this warning at: https://sqlalche.me/e/20/cprf)\n", + " result = connection.execute(text(query))\n" + ] + }, + { + "data": { + "text/html": [ + "
\n", + "\n", + "\n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + " \n", + "
namekafka.topictagvaluetimestampinserted_at
0Counterraw_scouter-opcua-orchestrated-pipelinens=2;i=2-50.1322025-07-02 13:05:192025-07-02 13:05:19.729000
1Rolloutraw_scouter-opcua-orchestrated-pipelinens=2;i=370.7672025-07-02 13:05:192025-07-02 13:05:19.731000
2Squareraw_scouter-opcua-orchestrated-pipelinens=2;i=4-58.4482025-07-02 13:05:192025-07-02 13:05:19.732000
3Counterraw_scouter-opcua-orchestrated-pipelinens=2;i=2-50.1262025-07-02 13:05:242025-07-02 13:05:24.728000
4Rolloutraw_scouter-opcua-orchestrated-pipelinens=2;i=369.1992025-07-02 13:05:242025-07-02 13:05:24.730000
.....................
373Rolloutraw_scouter-opcua-orchestrated-pipelinens=2;i=385.9212025-07-02 13:15:402025-07-02 13:15:40.230000
374Squareraw_scouter-opcua-orchestrated-pipelinens=2;i=4-69.2962025-07-02 13:15:402025-07-02 13:15:40.232000
375Counterraw_scouter-opcua-orchestrated-pipelinens=2;i=2-71.2072025-07-02 13:15:452025-07-02 13:15:45.228000
376Rolloutraw_scouter-opcua-orchestrated-pipelinens=2;i=384.6652025-07-02 13:15:452025-07-02 13:15:45.231000
377Squareraw_scouter-opcua-orchestrated-pipelinens=2;i=4-67.5922025-07-02 13:15:452025-07-02 13:15:45.233000
\n", + "

378 rows × 6 columns

\n", + "
" + ], + "text/plain": [ + " name kafka.topic tag value \\\n", + "0 Counter raw_scouter-opcua-orchestrated-pipeline ns=2;i=2 -50.132 \n", + "1 Rollout raw_scouter-opcua-orchestrated-pipeline ns=2;i=3 70.767 \n", + "2 Square raw_scouter-opcua-orchestrated-pipeline ns=2;i=4 -58.448 \n", + "3 Counter raw_scouter-opcua-orchestrated-pipeline ns=2;i=2 -50.126 \n", + "4 Rollout raw_scouter-opcua-orchestrated-pipeline ns=2;i=3 69.199 \n", + ".. ... ... ... ... \n", + "373 Rollout raw_scouter-opcua-orchestrated-pipeline ns=2;i=3 85.921 \n", + "374 Square raw_scouter-opcua-orchestrated-pipeline ns=2;i=4 -69.296 \n", + "375 Counter raw_scouter-opcua-orchestrated-pipeline ns=2;i=2 -71.207 \n", + "376 Rollout raw_scouter-opcua-orchestrated-pipeline ns=2;i=3 84.665 \n", + "377 Square raw_scouter-opcua-orchestrated-pipeline ns=2;i=4 -67.592 \n", + "\n", + " timestamp inserted_at \n", + "0 2025-07-02 13:05:19 2025-07-02 13:05:19.729000 \n", + "1 2025-07-02 13:05:19 2025-07-02 13:05:19.731000 \n", + "2 2025-07-02 13:05:19 2025-07-02 13:05:19.732000 \n", + "3 2025-07-02 13:05:24 2025-07-02 13:05:24.728000 \n", + "4 2025-07-02 13:05:24 2025-07-02 13:05:24.730000 \n", + ".. ... ... \n", + "373 2025-07-02 13:15:40 2025-07-02 13:15:40.230000 \n", + "374 2025-07-02 13:15:40 2025-07-02 13:15:40.232000 \n", + "375 2025-07-02 13:15:45 2025-07-02 13:15:45.228000 \n", + "376 2025-07-02 13:15:45 2025-07-02 13:15:45.231000 \n", + "377 2025-07-02 13:15:45 2025-07-02 13:15:45.233000 \n", + "\n", + "[378 rows x 6 columns]" + ] + }, + "metadata": {}, + "output_type": "display_data" + } + ], + "source": [ + "from sqlalchemy import create_engine, MetaData, Table, select, func, text\n", + "import pandas as pd\n", + "from datetime import datetime\n", + "\n", + "engine = create_engine('druid://localhost:8082/druid/v2/sql/')\n", + "metadata = MetaData()\n", + "places = Table('raw_scouter-opcua-orchestrated-pipeline', metadata, autoload_with=engine)\n", + "date_str = '2025-01-01'\n", + "stmt = select(places).where(text(f'\"__time\" > TIMESTAMP \\'{date_str}\\''))\n", + "\n", + "result = pd.read_sql(stmt, engine)\n", + "\n", + "result[\"inserted_at\"] = pd.to_datetime(result[\"__time\"]).dt.strftime(\n", + " \"%Y-%m-%d %H:%M:%S.%f\")\n", + "\n", + "result.drop(columns=[\"__time\"], inplace=True)\n", + "\n", + "display(result)" + ] + } + ], + "metadata": { + "kernelspec": { + "display_name": "venv", + "language": "python", + "name": "python3" + }, + "language_info": { + "codemirror_mode": { + "name": "ipython", + "version": 3 + }, + "file_extension": ".py", + "mimetype": "text/x-python", + "name": "python", + "nbconvert_exporter": "python", + "pygments_lexer": "ipython3", + "version": "3.11.13" + } + }, + "nbformat": 4, + "nbformat_minor": 2 +} diff --git a/values.yaml b/values.yaml index 589a8c7..884ccf7 100644 --- a/values.yaml +++ b/values.yaml @@ -182,6 +182,11 @@ env: - name: MONGODB_DATABASE value: "sientia" + - name: DRUID_HOST + value: "druid-router.druid.svc.cluster.local" + - name: DRUID_PORT + value: "8081" + ssh: enabled: true secretName: git-ssh-key-sientia-scouter-worker