airflow.providers.amazon.aws.sensors.sqs¶

從 SQS 佇列中讀取訊息，然後刪除它。

類¶

SqsSensor

從 Amazon SQS 佇列獲取訊息，然後從佇列中刪除這些訊息。

模組內容¶

class airflow.providers.amazon.aws.sensors.sqs.SqsSensor(*, sqs_queue, max_messages=5, num_batches=1, wait_time_seconds=1, visibility_timeout=None, message_filtering=None, message_filtering_match_values=None, message_filtering_config=None, delete_message_on_reception=True, deferrable=conf.getboolean('operators', 'default_deferrable', fallback=False), **kwargs)[source]¶

基類: airflow.providers.amazon.aws.sensors.base_aws.AwsBaseSensor[airflow.providers.amazon.aws.hooks.sqs.SqsHook]

從 Amazon SQS 佇列獲取訊息，然後從佇列中刪除這些訊息。

如果訊息刪除失敗，則丟擲 AirflowException。否則，訊息將透過 XCom 推送，鍵為 messages。

預設情況下，感測器每次 poke 只執行一次 SQS 呼叫，這將結果限制為最多 10 條訊息。然而，每次 poke 的 SQS API 呼叫總數可以透過 num_batches 引數控制。

另請參閱

有關如何使用此感測器的更多資訊，請檢視指南：從 Amazon SQS 佇列讀取訊息

引數:

sqs_queue – SQS 佇列 URL (支援模板)
max_messages (int) – 每次 poke 檢索的最大訊息數 (支援模板)
num_batches (int) – 感測器呼叫 SQS API 接收訊息的次數 (預設值: 1)
wait_time_seconds (int) – 等待接收訊息的時間（秒）(預設值: 1 秒)
visibility_timeout (int | None) – 可見性超時，在此期間 Amazon SQS 會阻止其他消費者接收和處理該訊息。
message_filtering (airflow.providers.amazon.aws.utils.sqs.MessageFilteringType | None) – 指定如何過濾收到的訊息。支援的選項有：None (不過濾，預設值)、‘literal’ (訊息體精確匹配)、‘jsonpath’ (使用 JSONPath 表示式過濾訊息體)，或 ‘jsonpath-ext’ (類似於 ‘jsonpath’，但帶有擴充套件的查詢語法)。您可以透過覆蓋相關的類方法來新增更多方法。
message_filtering_match_values (Any) – 訊息過濾器的可選匹配值。例如，對於精確匹配，如果訊息體與任何指定的值匹配，則包含該訊息。對於 JSONPath 匹配，使用 JSONPath 表示式的結果，如果與任何指定的值匹配，則包含該訊息。
message_filtering_config (Any) – 傳遞給訊息過濾器的附加配置。例如，對於 JSONPath 過濾，您可以在此處傳遞 JSONPath 表示式字串，例如 ‘foo[*].baz’。訊息體不匹配的訊息將被忽略。
delete_message_on_reception (bool) – 預設為 True，訊息一經消費即從佇列中刪除。否則，訊息消費後仍保留在佇列中，需要手動刪除。
deferrable (bool) – 如果為 True，感測器將以可推遲模式執行。此模式需要安裝 aiobotocore 模組。(預設值: False，但可以透過在配置檔案中設定 default_deferrable 為 True 來覆蓋)
aws_conn_id – 用於 AWS 憑據的 Airflow 連線。如果此引數為 None 或為空，則使用預設的 boto3 行為。如果 Airflow 以分散式方式執行且 aws_conn_id 為 None 或為空，則將使用預設的 boto3 配置（且必須在每個工作節點上維護）。
region_name – AWS region_name。如果未指定，則使用預設的 boto3 行為。
verify – 是否驗證 SSL 證書。詳見：https://boto3.amazonaws.com/v1/documentation/api/latest/reference/core/session.html
botocore_config – botocore 客戶端的配置字典（鍵值對）。詳見：https://botocore.amazonaws.com/v1/documentation/api/latest/reference/config.html

aws_hook_class[source]¶

template_fields: collections.abc.Sequence[str][source]¶

sqs_queue[source]¶

max_messages = 5[source]¶

num_batches = 1[source]¶

wait_time_seconds = 1[source]¶

visibility_timeout = None[source]¶

message_filtering = None[source]¶

delete_message_on_reception = True[source]¶

message_filtering_match_values = None[source]¶

message_filtering_config = None[source]¶

deferrable = True[source]¶

execute(context)[source]¶

建立運算子時派生。

執行任務的主要方法。Context 是渲染 jinja 模板時使用的相同字典。

有關更多上下文資訊，請參閱 get_template_context。

execute_complete(context, event=None)[source]¶

poll_sqs(sqs_conn)[source]¶

輪詢 SQS 佇列以檢索訊息。

引數:: sqs_conn (airflow.providers.amazon.aws.hooks.base_aws.BaseAwsConnection) – SQS 連線
返回:: 從 SQS 檢索到的訊息列表
返回型別:: collections.abc.Collection

poke(context)[source]¶

檢查訂閱佇列中的訊息，並使用 messages 鍵將其寫入 xcom。

引數:: context (airflow.utils.context.Context) – 上下文物件
返回:: 如果訊息可用則為 True，否則為 False