This is an automated email from the ASF dual-hosted git repository.
gopidesu pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 6589798823f Update apache Cassandra and Kylin examples (#52497)
6589798823f is described below
commit 6589798823f15d783cda0c59c60c752dde3b91b3
Author: GPK <[email protected]>
AuthorDate: Tue Aug 19 20:50:57 2025 +0100
Update apache Cassandra and Kylin examples (#52497)
* Update apache providers examples
* Update apache providers examples
* Resolve comments
---
.../apache/cassandra/example_cassandra_dag.py | 9 ++++--
.../tests/system/apache/kylin/example_kylin_dag.py | 32 +++++++++++++++-------
2 files changed, 29 insertions(+), 12 deletions(-)
diff --git
a/providers/apache/cassandra/tests/system/apache/cassandra/example_cassandra_dag.py
b/providers/apache/cassandra/tests/system/apache/cassandra/example_cassandra_dag.py
index 5c1fe9b5271..b2d98d88df4 100644
---
a/providers/apache/cassandra/tests/system/apache/cassandra/example_cassandra_dag.py
+++
b/providers/apache/cassandra/tests/system/apache/cassandra/example_cassandra_dag.py
@@ -45,9 +45,14 @@ with DAG(
catchup=False,
tags=["example"],
) as dag:
- table_sensor = CassandraTableSensor(task_id="cassandra_table_sensor")
+ # Replace <table_name> with your actual table name
+
+ table_sensor = CassandraTableSensor(task_id="cassandra_table_sensor",
table="<table_name>")
+
+ record_sensor = CassandraRecordSensor(
+ task_id="cassandra_record_sensor", keys={"p1": "v1", "p2": "v2"},
table="<table_name>"
+ )
- record_sensor = CassandraRecordSensor(task_id="cassandra_record_sensor",
keys={"p1": "v1", "p2": "v2"})
# [END howto_operator_cassandra_sensors]
from tests_common.test_utils.system_tests import get_test_run # noqa: E402
diff --git
a/providers/apache/kylin/tests/system/apache/kylin/example_kylin_dag.py
b/providers/apache/kylin/tests/system/apache/kylin/example_kylin_dag.py
index e4de4105e05..cf5833654b9 100644
--- a/providers/apache/kylin/tests/system/apache/kylin/example_kylin_dag.py
+++ b/providers/apache/kylin/tests/system/apache/kylin/example_kylin_dag.py
@@ -28,6 +28,11 @@ from datetime import datetime
from airflow import DAG
from airflow.providers.apache.kylin.operators.kylin_cube import
KylinCubeOperator
+try:
+ from airflow.sdk import chain
+except ImportError:
+ from airflow.models.baseoperator import chain # type: ignore[no-redef]
+
ENV_ID = os.environ.get("SYSTEM_TESTS_ENV_ID")
DAG_ID = "example_kylin_operator"
@@ -49,21 +54,19 @@ with DAG(
return {"date_start": "1325347200000", "date_end": "1325433600000"}
gen_build_time_task = gen_build_time()
- gen_build_time_output_date_start = gen_build_time_task["date_start"]
- gen_build_time_output_date_end = gen_build_time_task["date_end"]
build_task1 = KylinCubeOperator(
task_id="kylin_build_1",
command="build",
- start_time=gen_build_time_output_date_start,
- end_time=gen_build_time_output_date_end,
+ start_time="{{
task_instance.xcom_pull(task_ids='gen_build_time')['date_start'] }}",
+ end_time="{{
task_instance.xcom_pull(task_ids='gen_build_time')['date_end'] }}",
is_track_job=True,
)
build_task2 = KylinCubeOperator(
task_id="kylin_build_2",
command="build",
- start_time=gen_build_time_output_date_end,
+ start_time="{{
task_instance.xcom_pull(task_ids='gen_build_time')['date_start'] }}",
end_time="1325520000000",
is_track_job=True,
)
@@ -71,15 +74,15 @@ with DAG(
refresh_task1 = KylinCubeOperator(
task_id="kylin_refresh_1",
command="refresh",
- start_time=gen_build_time_output_date_start,
- end_time=gen_build_time_output_date_end,
+ start_time="{{
task_instance.xcom_pull(task_ids='gen_build_time')['date_start'] }}",
+ end_time="{{
task_instance.xcom_pull(task_ids='gen_build_time')['date_end'] }}",
is_track_job=True,
)
merge_task = KylinCubeOperator(
task_id="kylin_merge",
command="merge",
- start_time=gen_build_time_output_date_start,
+ start_time="{{
task_instance.xcom_pull(task_ids='gen_build_time')['date_start'] }}",
end_time="1325520000000",
is_track_job=True,
)
@@ -97,11 +100,20 @@ with DAG(
build_task3 = KylinCubeOperator(
task_id="kylin_build_3",
command="build",
- start_time=gen_build_time_output_date_end,
+ start_time="{{
task_instance.xcom_pull(task_ids='gen_build_time')['date_start'] }}",
end_time="1328730000000",
)
- build_task1 >> build_task2 >> refresh_task1 >> merge_task >> disable_task
>> purge_task >> build_task3
+ chain(
+ gen_build_time_task,
+ build_task1,
+ build_task2,
+ refresh_task1,
+ merge_task,
+ disable_task,
+ purge_task,
+ build_task3,
+ )
# Task dependency created via `XComArgs`:
# gen_build_time >> build_task1
