Add howto docs for Cassandra Operators (issues#8189)

.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

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.

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.record import CassandraRecordSensor
+from airflow.providers.apache.cassandra.sensors.table import CassandraTableSensor
+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]

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:

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:
 

docs/howto/connection/cassandra.rst

@@ -0,0 +1,114 @@
+ .. 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 <https://cassandra.apache.org/>`__.
+
+Default Connection IDs
+----------------------
+
+Cassandra hook and Cassandra operators use ``cassandra_default`` by default.
+
+Configuring the Connection
+--------------------------
+Host (required)
+    The host to connect to. It is possible to specify multiple hosts as a comma-separated list.
+
+Schema (required)
+    The schema (keyspace) name to be used in the database.
+
+Login (required)
+    The user name to connect.
+
+Password (required)
+    The password to connect.
+
+Port (required)
+    The port to connect.
+
+Extra (optional)
+    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``. Following are a few samples 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/

docs/howto/operator/apache/cassandra.rst

@@ -0,0 +1,70 @@
+ .. 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
+==========================
+
+`Apache Cassandra <https://cassandra.apache.org/>`__ 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
+  :local:
+
+Prerequisite
+------------
+
+To use operators, you must configure a :doc:`Cassandra Connection <../../connection/cassandra>`.
+
+.. _howto/operator:CassandraTableSensor:
+
+Waiting for a Table to be created
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+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.
+
+.. 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]
+
+
+.. _howto/operator:CassandraRecordSensor:
+
+Waiting for a Record to be created
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+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.
+
+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 <https://cassandra.apache.org/doc/latest/cql/dml.html#select>`_.

docs/operators-and-hooks-ref.rst

@@ -122,7 +122,7 @@ Foundation.
      - Sensor
 
    * - `Apache Cassandra <http://cassandra.apache.org/>`__
-     -
+     - :doc:`How to use <howto/operator/apache/cassandra>`
      - :mod:`airflow.providers.apache.cassandra.hooks.cassandra`
      -
      - :mod:`airflow.providers.apache.cassandra.sensors.record`,