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/__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..ec9b491e6f473 --- /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.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] 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 new file mode 100644 index 0000000000000..69146738eb476 --- /dev/null +++ b/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 `__. + +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/ diff --git a/docs/howto/operator/apache/cassandra.rst b/docs/howto/operator/apache/cassandra.rst new file mode 100644 index 0000000000000..ede4f89ed7ed4 --- /dev/null +++ b/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 `__ 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 `_. 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`,