apache-airflow-providers-sqream-blue
=====================================
Apache Airflow is a popular open source orchestration tool. It allows users to write complex workflows composed of multiple kinds of actions and services using DAGs, and to schedule, debug and monitor workflow runs.
Different kind of actions are represented with specialized Python classes, called “Operators”. Each SaaS database vendor can build one or more customized Operators for performing different DB operations (e.g. execute arbitrary SQL statements, schedule DB job runs etc.).
This package is an Operator for executing SQL statments on SQream Blue using Python connector.
Requirements
-------------
* Python 3.9+
Installing the Airflow-provider-sqream-blue
-------------------------------------------
The Airflow provider sqream-blue is available via `PyPi <https://pypi.org/project/airflow-provider-sqream-blue/>`_.
Install the connector with ``pip3``:
.. code-block:: console
pip3 install airflow-provider-sqream-blue
``pip3`` will automatically installs all necessary libraries and modules.
How to use the Airflow-provider-sqream-blue
-------------------------------------------
Create a connection -
After the installation of the package on the Airflow server,
refresh the server and create a new connection.
sqream-blue will apper on the connection-type.
.. image:: images/create_connection.png
:width: 800
Click test and save after enter connection params.
Create a dag -
Create a python dag file and copy it to dags folder on the airflow server -
To find dag folder run this command
.. code-block:: console
airflow config list | grep dags_folder
Example of python dag file
.. code-block:: python
import logging
from datetime import timedelta
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from sqream_blue.operators.sqream_blue import SQreamBlueSqlOperator
from sqream_blue.hooks.sqream_blue import SQreamBlueHook
from airflow.utils.dates import days_ago
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
with DAG(
dag_id='Test_Dag',
schedule_interval='0 0 * * *',
start_date=days_ago(2),
dagrun_timeout=timedelta(minutes=60),
template_searchpath=['/home/sqream/'],
tags=['Test']
) as dag:
list_operator = SQreamBlueSqlOperator(
task_id='create_and_insert',
sql=['create or replace table t_a(x int not null)', 'insert into t_a values (1)', 'insert into t_a values (2)'],
sqream_blue_conn_id="sqream_blue_connection",
dag=dag,
)
simple_operator = SQreamBlueSqlOperator(
task_id='just_select',
sql='select * from t_a',
sqream_blue_conn_id="sqream_blue_connection",
dag=dag,
)
sql_file_operator = SQreamBlueSqlOperator(
task_id='sql_file',
sql='daniel.sql',
sqream_blue_conn_id="sqream_blue_connection",
dag=dag,
)
def count_python(**context):
dwh_hook = SQreamBlueHook(sqream_blue_conn_id="sqream_blue_connection")
result = dwh_hook.get_first("select count(*) from public.t_a")
logging.info("Number of rows in `public.t_a` - %s", result[0])
count_through_python_operator_query = PythonOperator(
task_id="log_row_count",
python_callable=count_python)
list_operator >> simple_operator >> count_through_python_operator_query >> sql_file_operator
The execution of the Dag File -
.. image:: images/execution_dag.png
:width: 600
Raw data
{
"_id": null,
"home_page": "https://github.com/SQream/apache-airflow-providers-sqream-blue",
"name": "airflow-provider-sqream-blue",
"maintainer": "",
"docs_url": null,
"requires_python": ">=3.9",
"maintainer_email": "",
"keywords": "",
"author": "SQream",
"author_email": "info@sqream.com",
"download_url": "https://files.pythonhosted.org/packages/a8/4b/10f61e9d47b132647546407830da3908650e30020ecf972a9b85425c7498/airflow-provider-sqream-blue-0.0.13.tar.gz",
"platform": null,
"description": "apache-airflow-providers-sqream-blue\n=====================================\nApache Airflow is a popular open source orchestration tool. It allows users to write complex workflows composed of multiple kinds of actions and services using DAGs, and to schedule, debug and monitor workflow runs.\n\nDifferent kind of actions are represented with specialized Python classes, called \u201cOperators\u201d. Each SaaS database vendor can build one or more customized Operators for performing different DB operations (e.g. execute arbitrary SQL statements, schedule DB job runs etc.).\n\nThis package is an Operator for executing SQL statments on SQream Blue using Python connector.\n\n\nRequirements\n-------------\n\n* Python 3.9+\n\n\nInstalling the Airflow-provider-sqream-blue\n-------------------------------------------\nThe Airflow provider sqream-blue is available via `PyPi <https://pypi.org/project/airflow-provider-sqream-blue/>`_.\n\nInstall the connector with ``pip3``:\n\n.. code-block:: console\n\n pip3 install airflow-provider-sqream-blue\n\n``pip3`` will automatically installs all necessary libraries and modules.\n\n\nHow to use the Airflow-provider-sqream-blue\n-------------------------------------------\nCreate a connection - \n\nAfter the installation of the package on the Airflow server, \nrefresh the server and create a new connection.\nsqream-blue will apper on the connection-type.\n\n.. image:: images/create_connection.png\n :width: 800\n\nClick test and save after enter connection params.\n\nCreate a dag - \n \nCreate a python dag file and copy it to dags folder on the airflow server -\n \nTo find dag folder run this command\n\n.. code-block:: console\n \n airflow config list | grep dags_folder\n\nExample of python dag file\n\n.. code-block:: python\n\n import logging\n from datetime import timedelta\n from airflow import DAG\n from airflow.operators.python_operator import PythonOperator\n from sqream_blue.operators.sqream_blue import SQreamBlueSqlOperator\n from sqream_blue.hooks.sqream_blue import SQreamBlueHook\n from airflow.utils.dates import days_ago\n\n logging.basicConfig(level=logging.INFO)\n logger = logging.getLogger(__name__)\n\n with DAG(\n dag_id='Test_Dag',\n schedule_interval='0 0 * * *',\n start_date=days_ago(2),\n dagrun_timeout=timedelta(minutes=60),\n template_searchpath=['/home/sqream/'],\n tags=['Test']\n ) as dag:\n\n list_operator = SQreamBlueSqlOperator(\n task_id='create_and_insert',\n sql=['create or replace table t_a(x int not null)', 'insert into t_a values (1)', 'insert into t_a values (2)'],\n sqream_blue_conn_id=\"sqream_blue_connection\",\n dag=dag,\n )\n\n simple_operator = SQreamBlueSqlOperator(\n task_id='just_select',\n sql='select * from t_a',\n sqream_blue_conn_id=\"sqream_blue_connection\",\n dag=dag,\n )\n\n sql_file_operator = SQreamBlueSqlOperator(\n task_id='sql_file',\n sql='daniel.sql',\n sqream_blue_conn_id=\"sqream_blue_connection\",\n dag=dag,\n )\n\n def count_python(**context):\n dwh_hook = SQreamBlueHook(sqream_blue_conn_id=\"sqream_blue_connection\")\n result = dwh_hook.get_first(\"select count(*) from public.t_a\")\n logging.info(\"Number of rows in `public.t_a` - %s\", result[0])\n\n count_through_python_operator_query = PythonOperator(\n task_id=\"log_row_count\",\n python_callable=count_python)\n\n\n list_operator >> simple_operator >> count_through_python_operator_query >> sql_file_operator\n\n\nThe execution of the Dag File -\n\n.. image:: images/execution_dag.png\n :width: 600\n \n",
"bugtrack_url": null,
"license": "Apache License 2.0",
"summary": "About Apache Airflow - A platform to programmatically author, schedule, and monitor workflows.",
"version": "0.0.13",
"project_urls": {
"Homepage": "https://github.com/SQream/apache-airflow-providers-sqream-blue"
},
"split_keywords": [],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "4c2530d62ff0f860a5f4c4b1a221d916687270b8347b2a2ce76b8822ca3dba64",
"md5": "85510dd7a65dbf7a7506e9444e836b8c",
"sha256": "5b21703f3539a6d6aea451013e2259259791364bf9e4b8bac3a0fad3dfc5314d"
},
"downloads": -1,
"filename": "airflow_provider_sqream_blue-0.0.13-py3-none-any.whl",
"has_sig": false,
"md5_digest": "85510dd7a65dbf7a7506e9444e836b8c",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.9",
"size": 8929,
"upload_time": "2023-11-22T08:48:26",
"upload_time_iso_8601": "2023-11-22T08:48:26.754922Z",
"url": "https://files.pythonhosted.org/packages/4c/25/30d62ff0f860a5f4c4b1a221d916687270b8347b2a2ce76b8822ca3dba64/airflow_provider_sqream_blue-0.0.13-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "a84b10f61e9d47b132647546407830da3908650e30020ecf972a9b85425c7498",
"md5": "80c6aa5aee015584332e844e40fa784c",
"sha256": "4b09755578fafdd36549adff0d358c1437281119d16a93f599248c86d2cc199f"
},
"downloads": -1,
"filename": "airflow-provider-sqream-blue-0.0.13.tar.gz",
"has_sig": false,
"md5_digest": "80c6aa5aee015584332e844e40fa784c",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.9",
"size": 7611,
"upload_time": "2023-11-22T08:48:28",
"upload_time_iso_8601": "2023-11-22T08:48:28.722623Z",
"url": "https://files.pythonhosted.org/packages/a8/4b/10f61e9d47b132647546407830da3908650e30020ecf972a9b85425c7498/airflow-provider-sqream-blue-0.0.13.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2023-11-22 08:48:28",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "SQream",
"github_project": "apache-airflow-providers-sqream-blue",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"requirements": [],
"lcname": "airflow-provider-sqream-blue"
}
JSON
