From 77e046cc9e423dc81be1fc8eb4f898e7597764c6 Mon Sep 17 00:00:00 2001 From: rootcss Date: Sun, 19 Jul 2020 01:25:47 +0530 Subject: [PATCH 1/2] add howto docs for Cassandra Operators (cherry picked from commit 456e4a4988b9cb614c22c3ca3577522472e50b7b) --- .../apache/cassandra/example_dags/__init__.py | 16 +++ .../example_dags/example_cassandra_dag.py | 54 +++++++++ docs/howto/connection/cassandra.rst | 109 ++++++++++++++++++ docs/howto/operator/apache/cassandra.rst | 71 ++++++++++++ docs/operators-and-hooks-ref.rst | 2 +- 5 files changed, 251 insertions(+), 1 deletion(-) create mode 100644 airflow/providers/apache/cassandra/example_dags/__init__.py create mode 100644 airflow/providers/apache/cassandra/example_dags/example_cassandra_dag.py create mode 100644 docs/howto/connection/cassandra.rst create mode 100644 docs/howto/operator/apache/cassandra.rst diff --git a/airflow/providers/apache/cassandra/example_dags/__init__.py b/airflow/providers/apache/cassandra/example_dags/__init__.py new file mode 100644 index 0000000000000..13a83393a9124 --- /dev/null +++ b/airflow/providers/apache/cassandra/example_dags/__init__.py @@ -0,0 +1,16 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. diff --git a/airflow/providers/apache/cassandra/example_dags/example_cassandra_dag.py b/airflow/providers/apache/cassandra/example_dags/example_cassandra_dag.py new file mode 100644 index 0000000000000..4a1889ea918f8 --- /dev/null +++ b/airflow/providers/apache/cassandra/example_dags/example_cassandra_dag.py @@ -0,0 +1,54 @@ +# +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. + +""" +Example Airflow DAG to check if a Cassandra Table and a Records exists +or not using `CassandraTableSensor` and `CassandraRecordSensor`. +""" +from airflow.models import DAG +from airflow.providers.apache.cassandra.sensors.table import CassandraTableSensor +from airflow.providers.apache.cassandra.sensors.record import CassandraRecordSensor +from airflow.utils.dates import days_ago + +args = { + 'owner': 'Airflow', + 'start_date': days_ago(2) +} + +with DAG( + dag_id='example_cassandra_operator', + default_args=args, + schedule_interval=None, + tags=['example'] +) as dag: + # [START howto_operator_cassandra_table_sensor] + table_sensor = CassandraTableSensor( + task_id="cassandra_table_sensor", + cassandra_conn_id="cassandra_default", + table="keyspace_name.table_name", + ) + # [END howto_operator_cassandra_table_sensor] + + # [START howto_operator_cassandra_record_sensor] + record_sensor = CassandraRecordSensor( + task_id="cassandra_record_sensor", + cassandra_conn_id="cassandra_default", + table="keyspace_name.table_name", + keys={"p1": "v1", "p2": "v2"}, + ) + # [END howto_operator_cassandra_record_sensor] diff --git a/docs/howto/connection/cassandra.rst b/docs/howto/connection/cassandra.rst new file mode 100644 index 0000000000000..683a74078c392 --- /dev/null +++ b/docs/howto/connection/cassandra.rst @@ -0,0 +1,109 @@ + .. Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance +with the License. You may obtain a copy of the License at + + .. http://www.apache.org/licenses/LICENSE-2.0 + + .. Unless required by applicable law or agreed to in writing, +software distributed under the License is distributed on an +"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +KIND, either express or implied. See the License for the +specific language governing permissions and limitations +under the License. + +Apache Cassandra Connection +======================= + +The Apache Cassandra connection type enables connection to Apache Cassandra. + +Default Connection IDs +---------------------- + +Cassandra hook and Cassandra operators use ``cassandra_default`` by default. + +Configuring the Connection +-------------------------- +Host (required) + The host to connect to. Hosts can be provided in comma-separated string values. + +Schema (required) + Specify the schema (keyspace) name to be used in the database. + +Login (required) + Specify the user name to connect. + +Password (required) + Specify the password to connect. + +Port (required) + Specify the port to connect. + +Extra (optional) + Specify the extra parameters (as json dictionary) that can be used in cassandra + connection. The following parameters out of the standard python parameters + are supported: + + * ``load_balancing_policy`` - This parameter specifies the load balancing policy to be used. There are four available policies: + ``RoundRobinPolicy``, ``DCAwareRoundRobinPolicy``, ``WhiteListRoundRobinPolicy`` and ``TokenAwarePolicy``. ``RoundRobinPolicy`` is the default load balancing policy. + * ``load_balancing_policy_args`` - This parameter specifies the arguments for the load balancing policy being used. + * ``cql_version`` - This parameter specifies the CQL version of cassandra. + * ``ssl_options`` - This parameter specifies the details related to SSL, if it's enabled in Cassandra. + + +Examples for the **Extra** field +-------------------------------- + +1. Specifying ``ssl_options`` + +If SSL is enabled in Cassandra, pass in a dict in the extra field as kwargs for ``ssl.wrap_socket()``. For example:: + +.. code-block:: json + + { + "ssl_options" : { + "ca_certs" : "PATH/TO/CA_CERTS" + } + } + + +2. Specifying ``load_balancing_policy`` and ``load_balancing_policy_args`` + +Default load balancing policy is ``RoundRobinPolicy``. To specify a different LB policy:: + +* DCAwareRoundRobinPolicy + +.. code-block:: json + { + "load_balancing_policy": "DCAwareRoundRobinPolicy", + "load_balancing_policy_args": { + "local_dc": "LOCAL_DC_NAME", + "used_hosts_per_remote_dc": "SOME_INT_VALUE", + } + } + +* WhiteListRoundRobinPolicy + +.. code-block:: json + { + "load_balancing_policy": "WhiteListRoundRobinPolicy", + "load_balancing_policy_args": { + "hosts": ["HOST1", "HOST2", "HOST3"] + } + } +* TokenAwarePolicy + +.. code-block:: json + { + "load_balancing_policy": "TokenAwarePolicy", + "load_balancing_policy_args": { + "child_load_balancing_policy": "CHILD_POLICY_NAME", + "child_load_balancing_policy_args": {} + } + } + +.. seealso:: + https://pypi.org/project/cassandra-driver/ diff --git a/docs/howto/operator/apache/cassandra.rst b/docs/howto/operator/apache/cassandra.rst new file mode 100644 index 0000000000000..ac4b2798aa35b --- /dev/null +++ b/docs/howto/operator/apache/cassandra.rst @@ -0,0 +1,71 @@ + .. Licensed to the Apache Software Foundation (ASF) under one +or more contributor license agreements. See the NOTICE file +distributed with this work for additional information +regarding copyright ownership. The ASF licenses this file +to you under the Apache License, Version 2.0 (the +"License"); you may not use this file except in compliance +with the License. You may obtain a copy of the License at + + .. http://www.apache.org/licenses/LICENSE-2.0 + + .. Unless required by applicable law or agreed to in writing, +software distributed under the License is distributed on an +"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +KIND, either express or implied. See the License for the +specific language governing permissions and limitations +under the License. + + +Apache Cassandra Operators +====================== + +.. contents:: + :depth: 1 + :local: + +Prerequisite +------------ + +To use ``CassandraTableSensor`` and ``CassandraRecordSensor``, you must configure a :doc:`Cassandra Connection <../../connection/cassandra>`. + +:ref:`howto/operator:CassandraTableSensor` + +CassandraTableSensor +----------------- + +Checks for the existence of a table in a Cassandra cluster. + +For parameter definition take a look at :class:`~airflow.providers.apache.cassandra.sensors.table.CassandraTableSensor`. + +Use the ``table`` parameter to poke until the provided table is found. Use dot notation to target a specific keyspace. + +.. exampleinclude:: /../airflow/providers/apache/cassandra/example_dags/example_cassandra_dag.py + :language: python + :dedent: 4 + :start-after: [START howto_operator_cassandra_table_sensor] + :end-before: [END howto_operator_cassandra_table_sensor] + + +:ref:`howto/operator:CassandraRecordSensor` + +CassandraRecordSensor +----------------- + +Checks for the existence of a record in a Cassandra cluster. + +For parameter definition take a look at :class:`~airflow.providers.apache.cassandra.sensors.record.CassandraRecordSensor`. + +Use the ``table`` parameter to mention the keyspace and table for the record. Use dot notation to target a specific keyspace. + +Use the ``keys`` parameter to poke until the provided record is found. The existence of record is identified using key value pairs. In the given example, we're are looking for value ``v1`` in column ``p1`` and ``v2`` in column ``p2``. + +.. exampleinclude:: /../airflow/providers/apache/cassandra/example_dags/example_cassandra_dag.py + :language: python + :dedent: 4 + :start-after: [START howto_operator_cassandra_record_sensor] + :end-before: [END howto_operator_cassandra_record_sensor] + +Reference +""""""""" + +For further information, look at `Cassandra Query Language (CQL) SELECT statement `_. diff --git a/docs/operators-and-hooks-ref.rst b/docs/operators-and-hooks-ref.rst index 985e4a96647b4..6f5028f8d86e1 100644 --- a/docs/operators-and-hooks-ref.rst +++ b/docs/operators-and-hooks-ref.rst @@ -122,7 +122,7 @@ Foundation. - Sensor * - `Apache Cassandra `__ - - + - :doc:`How to use ` - :mod:`airflow.providers.apache.cassandra.hooks.cassandra` - - :mod:`airflow.providers.apache.cassandra.sensors.record`, From ab1726a26ca0a125b3c5ff145529fb0f065a434a Mon Sep 17 00:00:00 2001 From: Shekhar Singh Date: Sun, 19 Jul 2020 11:32:47 +0530 Subject: [PATCH 2/2] Fix bugs in cassandra operators howto guide MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: Kamil BreguĊ‚a --- .pre-commit-config.yaml | 1 + .../example_dags/example_cassandra_dag.py | 2 +- .../apache/cassandra/sensors/record.py | 4 ++ .../apache/cassandra/sensors/table.py | 5 ++ docs/howto/connection/cassandra.rst | 65 ++++++++++--------- docs/howto/operator/apache/cassandra.rst | 51 +++++++-------- 6 files changed, 71 insertions(+), 57 deletions(-) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 011c72824a4db..b7bacaeafaa44 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -270,6 +270,7 @@ repos: ^airflow/providers/apache/hive/PROVIDERS_CHANGES_*| ^airflow/providers/apache/hive/README.md$| ^tests/providers/apache/cassandra/hooks/test_cassandra.py$| + ^docs/howto/connection/cassandra.rst$| ^CHANGELOG.txt$ - id: consistent-pylint language: pygrep diff --git a/airflow/providers/apache/cassandra/example_dags/example_cassandra_dag.py b/airflow/providers/apache/cassandra/example_dags/example_cassandra_dag.py index 4a1889ea918f8..ec9b491e6f473 100644 --- a/airflow/providers/apache/cassandra/example_dags/example_cassandra_dag.py +++ b/airflow/providers/apache/cassandra/example_dags/example_cassandra_dag.py @@ -21,8 +21,8 @@ or not using `CassandraTableSensor` and `CassandraRecordSensor`. """ from airflow.models import DAG -from airflow.providers.apache.cassandra.sensors.table import CassandraTableSensor from airflow.providers.apache.cassandra.sensors.record import CassandraRecordSensor +from airflow.providers.apache.cassandra.sensors.table import CassandraTableSensor from airflow.utils.dates import days_ago args = { diff --git a/airflow/providers/apache/cassandra/sensors/record.py b/airflow/providers/apache/cassandra/sensors/record.py index 268ca44523f97..59c43e5ce20c1 100644 --- a/airflow/providers/apache/cassandra/sensors/record.py +++ b/airflow/providers/apache/cassandra/sensors/record.py @@ -31,6 +31,10 @@ class CassandraRecordSensor(BaseSensorOperator): """ Checks for the existence of a record in a Cassandra cluster. + .. seealso:: + For more information on how to use this operator, take a look at the guide: + :ref:`howto/operator:CassandraRecordSensor` + For example, if you want to wait for a record that has values 'v1' and 'v2' for each primary keys 'p1' and 'p2' to be populated in keyspace 'k' and table 't', instantiate it as follows: diff --git a/airflow/providers/apache/cassandra/sensors/table.py b/airflow/providers/apache/cassandra/sensors/table.py index 47b9679e989bc..7a0d42dfeb61f 100644 --- a/airflow/providers/apache/cassandra/sensors/table.py +++ b/airflow/providers/apache/cassandra/sensors/table.py @@ -32,6 +32,11 @@ class CassandraTableSensor(BaseSensorOperator): """ Checks for the existence of a table in a Cassandra cluster. + .. seealso:: + For more information on how to use this operator, take a look at the guide: + :ref:`howto/operator:CassandraTableSensor` + + For example, if you want to wait for a table called 't' to be created in a keyspace 'k', instantiate it as follows: diff --git a/docs/howto/connection/cassandra.rst b/docs/howto/connection/cassandra.rst index 683a74078c392..69146738eb476 100644 --- a/docs/howto/connection/cassandra.rst +++ b/docs/howto/connection/cassandra.rst @@ -1,24 +1,26 @@ .. Licensed to the Apache Software Foundation (ASF) under one -or more contributor license agreements. See the NOTICE file -distributed with this work for additional information -regarding copyright ownership. The ASF licenses this file -to you under the Apache License, Version 2.0 (the -"License"); you may not use this file except in compliance -with the License. You may obtain a copy of the License at + or more contributor license agreements. See the NOTICE file + distributed with this work for additional information + regarding copyright ownership. The ASF licenses this file + to you under the Apache License, Version 2.0 (the + "License"); you may not use this file except in compliance + with the License. You may obtain a copy of the License at .. http://www.apache.org/licenses/LICENSE-2.0 .. Unless required by applicable law or agreed to in writing, -software distributed under the License is distributed on an -"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -KIND, either express or implied. See the License for the -specific language governing permissions and limitations -under the License. + software distributed under the License is distributed on an + "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + KIND, either express or implied. See the License for the + specific language governing permissions and limitations + under the License. + + Apache Cassandra Connection -======================= +=========================== -The Apache Cassandra connection type enables connection to Apache Cassandra. +The Apache Cassandra connection type enables connection to `Apache Cassandra `__. Default Connection IDs ---------------------- @@ -28,22 +30,22 @@ Cassandra hook and Cassandra operators use ``cassandra_default`` by default. Configuring the Connection -------------------------- Host (required) - The host to connect to. Hosts can be provided in comma-separated string values. + The host to connect to. It is possible to specify multiple hosts as a comma-separated list. Schema (required) - Specify the schema (keyspace) name to be used in the database. + The schema (keyspace) name to be used in the database. Login (required) - Specify the user name to connect. + The user name to connect. Password (required) - Specify the password to connect. + The password to connect. Port (required) - Specify the port to connect. + The port to connect. Extra (optional) - Specify the extra parameters (as json dictionary) that can be used in cassandra + The extra parameters (as json dictionary) that can be used in cassandra connection. The following parameters out of the standard python parameters are supported: @@ -59,9 +61,9 @@ Examples for the **Extra** field 1. Specifying ``ssl_options`` -If SSL is enabled in Cassandra, pass in a dict in the extra field as kwargs for ``ssl.wrap_socket()``. For example:: +If SSL is enabled in Cassandra, pass in a dict in the extra field as kwargs for ``ssl.wrap_socket()``. For example: -.. code-block:: json +.. code-block:: JSON { "ssl_options" : { @@ -69,34 +71,37 @@ If SSL is enabled in Cassandra, pass in a dict in the extra field as kwargs for } } - 2. Specifying ``load_balancing_policy`` and ``load_balancing_policy_args`` -Default load balancing policy is ``RoundRobinPolicy``. To specify a different LB policy:: +Default load balancing policy is ``RoundRobinPolicy``. Following are a few samples to specify a different LB policy: + +DCAwareRoundRobinPolicy: -* DCAwareRoundRobinPolicy +.. code-block:: JSON -.. code-block:: json { "load_balancing_policy": "DCAwareRoundRobinPolicy", "load_balancing_policy_args": { "local_dc": "LOCAL_DC_NAME", - "used_hosts_per_remote_dc": "SOME_INT_VALUE", + "used_hosts_per_remote_dc": "SOME_INT_VALUE" } } -* WhiteListRoundRobinPolicy +WhiteListRoundRobinPolicy: + +.. code-block:: JSON -.. code-block:: json { "load_balancing_policy": "WhiteListRoundRobinPolicy", "load_balancing_policy_args": { "hosts": ["HOST1", "HOST2", "HOST3"] } } -* TokenAwarePolicy -.. code-block:: json +TokenAwarePolicy: + +.. code-block:: JSON + { "load_balancing_policy": "TokenAwarePolicy", "load_balancing_policy_args": { diff --git a/docs/howto/operator/apache/cassandra.rst b/docs/howto/operator/apache/cassandra.rst index ac4b2798aa35b..ede4f89ed7ed4 100644 --- a/docs/howto/operator/apache/cassandra.rst +++ b/docs/howto/operator/apache/cassandra.rst @@ -1,23 +1,26 @@ .. Licensed to the Apache Software Foundation (ASF) under one -or more contributor license agreements. See the NOTICE file -distributed with this work for additional information -regarding copyright ownership. The ASF licenses this file -to you under the Apache License, Version 2.0 (the -"License"); you may not use this file except in compliance -with the License. You may obtain a copy of the License at + or more contributor license agreements. See the NOTICE file + distributed with this work for additional information + regarding copyright ownership. The ASF licenses this file + to you under the Apache License, Version 2.0 (the + "License"); you may not use this file except in compliance + with the License. You may obtain a copy of the License at .. http://www.apache.org/licenses/LICENSE-2.0 .. Unless required by applicable law or agreed to in writing, -software distributed under the License is distributed on an -"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY -KIND, either express or implied. See the License for the -specific language governing permissions and limitations -under the License. + software distributed under the License is distributed on an + "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + KIND, either express or implied. See the License for the + specific language governing permissions and limitations + under the License. + Apache Cassandra Operators -====================== +========================== + +`Apache Cassandra `__ is an open source distributed NoSQL database that can be used when you need scalability and high availability without compromising performance. It offers linear scalability and fault-tolerance on commodity hardware or cloud infrastructure which makes it the perfect platform for mission-critical data. It supports multi-datacenter replication with lower latencies. .. contents:: :depth: 1 @@ -26,16 +29,14 @@ Apache Cassandra Operators Prerequisite ------------ -To use ``CassandraTableSensor`` and ``CassandraRecordSensor``, you must configure a :doc:`Cassandra Connection <../../connection/cassandra>`. +To use operators, you must configure a :doc:`Cassandra Connection <../../connection/cassandra>`. -:ref:`howto/operator:CassandraTableSensor` +.. _howto/operator:CassandraTableSensor: -CassandraTableSensor ------------------ +Waiting for a Table to be created +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ -Checks for the existence of a table in a Cassandra cluster. - -For parameter definition take a look at :class:`~airflow.providers.apache.cassandra.sensors.table.CassandraTableSensor`. +The :class:`~airflow.providers.apache.cassandra.sensors.table.CassandraTableSensor` operator is used to check for the existence of a table in a Cassandra cluster. Use the ``table`` parameter to poke until the provided table is found. Use dot notation to target a specific keyspace. @@ -46,14 +47,12 @@ Use the ``table`` parameter to poke until the provided table is found. Use dot n :end-before: [END howto_operator_cassandra_table_sensor] -:ref:`howto/operator:CassandraRecordSensor` - -CassandraRecordSensor ------------------ +.. _howto/operator:CassandraRecordSensor: -Checks for the existence of a record in a Cassandra cluster. +Waiting for a Record to be created +^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^ -For parameter definition take a look at :class:`~airflow.providers.apache.cassandra.sensors.record.CassandraRecordSensor`. +The :class:`~airflow.providers.apache.cassandra.sensors.record.CassandraRecordSensor` operator is used to check for the existence of a record of a table in the Cassandra cluster. Use the ``table`` parameter to mention the keyspace and table for the record. Use dot notation to target a specific keyspace. @@ -66,6 +65,6 @@ Use the ``keys`` parameter to poke until the provided record is found. The exist :end-before: [END howto_operator_cassandra_record_sensor] Reference -""""""""" +^^^^^^^^^ For further information, look at `Cassandra Query Language (CQL) SELECT statement `_.