YTsaurus on social media

<p>We&nbsp;have released the MVP of&nbsp;Apache Airflow Provider for YTsaurus. Currently, it&nbsp;includes operators for working with Cypress, can read and write tables, and run queries in&nbsp;Query Tracker.</p>

<p>To&nbsp;install Apache Airflow Provider see for <a href="https://ytsaurus.tech/docs/en/api/airflow/userdoc#ustanovka" target="_blank">instructions</a>.</p>

<p>Brief description of&nbsp;supported operators:</p>
<ul>
<li><code>ReadTableOperator</code>&nbsp;&mdash; read a&nbsp;YTsaurus table into XCom or&nbsp;S3</li>
<li><code>WriteTableOperator</code>&nbsp;&mdash; write a&nbsp;YTsaurus table from XCom or&nbsp;S3</li>
<li><code>RunQueryOperator</code>&nbsp;&mdash; execute a&nbsp;query in&nbsp;Query Tracker and write the result to&nbsp;XCom or&nbsp;S3</li>
<li><code>CreateOperator</code>&nbsp;&mdash; create a&nbsp;node in&nbsp;Cypress</li>
<li><code>SetOperator</code>&nbsp;&mdash; write a&nbsp;value to&nbsp;a&nbsp;node</li>
<li><code>GetOperator</code>&nbsp;&mdash; get node content and write it&nbsp;to&nbsp;XCom</li>
<li><code>RemoveOperator</code>&nbsp;&mdash; delete a&nbsp;node</li>
<li><code>ListOperator</code>&nbsp;&mdash; get a&nbsp;list of&nbsp;node children and write it&nbsp;to&nbsp;XCom</li>
</ul>
<p>More on&nbsp;each of&nbsp;the supported operators can be&nbsp;found in&nbsp;the <a href="https://ytsaurus.tech/docs/en/api/airflow/userdoc#operators" target="_blank">documentation</a>.</p>
<p>Example of&nbsp;a&nbsp;DAG that writes data from S3 to&nbsp;a&nbsp;cluster with subsequent processing (see the <a href="https://github.com/ytsaurus/ytsaurus-airflow-provider/tree/main/ytsaurus_airflow_provider/example_dags" target="_blank">repo</a> for more examples).</p>

    <div class="yfm-clipboard">
        <pre><code class="language-python"># Path to the object we need in S3.
# File contents:
# {"user_name": "Alice", "is_happy": true}
# {"user_name": "Bob", "is_happy": false}
# {"user_name": "Charlie", "is_happy": true}
# ...
users_data_s3 = ObjectStoragePath("s3://users-data/latest", conn_id="aws_default")

with DAG(
    "export_users_data_and_count_happy_users",
    start_date=datetime(2025, 3, 20, tzinfo=timezone.utc),
    schedule="@daily",
    tags=["yt"],
) as dag:
    # Load user data from S3 into a YTsaurus table
    # //home/users-data-{{ ds }}, where {{ ds }} - is the task execution date.
    # The operator also saves this path to XCom "path".
    upload_user_data_task = WriteTableOperator(
        task_id="upload_user_data",
        path="//home/users-data-{{ ds }}", # All operators support airflow templates reference
        object_storage_path=users_data_s3, # Path to the object in S3
        object_storage_format="json") # Format in which data is written in S3

    # Run a YQL query for the loaded table to count happy users.
    # The query result is saved to XCom "result_0".
    process_user_data_task = RunQueryOperator(
        task_id="process_user_data",
        query="PRAGMA yt.InferSchema = '1';\n"
              "SELECT COUNT_IF(is_happy) AS count FROM `{}`;".format(upload_user_data_task.output["path"]),
        engine="yql")

    # Write the query result to the table //home/happy-users-stat in a new row.
    write_happy_users_task = WriteTableOperator(
        task_id="write_happy_users",
        path="&lt;append=true&gt;//home/happy-users-stat",
        input_data=[{"date": "{{ ds }}", "happy_users": process_user_data_task.output["result_0"]}])

    upload_user_data_task &gt;&gt; process_user_data_task &gt;&gt; write_happy_users_task
</code></pre>

        <button class="yfm-clipboard-button">
            <svg width="16" height="16" viewBox="0 0 24 24" class="yfm-clipboard-icon" data-animation="51">
                <path fill="currentColor" d="M19,21H8V7H19M19,5H8A2,2 0 0,0 6,7V21A2,2 0 0,0 8,23H19A2,2 0 0,0 21,21V7A2,2 0 0,0 19,5M16,1H4A2,2 0 0,0 2,3V17H4V3H16V1Z"></path>
                <path stroke="currentColor" fill="transparent" stroke-width="1.5" d="M9.5 13l3 3l5 -5" visibility="hidden">
                    <animate id="visibileAnimation-51" attributeName="visibility" from="hidden" to="visible" dur="0.2s" fill="freeze" begin></animate>
                    <animate id="hideAnimation-51" attributeName="visibility" from="visible" to="hidden" dur="1s" begin="visibileAnimation-51.end+1" fill="freeze"></animate>
                </path>
            </svg>
        </button>
    </div>
<p>If&nbsp;you want to&nbsp;add something, please contact us&nbsp;via the <a href="https://t.me/ytsaurus" target="_blank">community chat</a> or&nbsp;create issues/PRs in&nbsp;the <a href="https://github.com/ytsaurus/ytsaurus-airflow-provider" target="_blank">repository</a>.</p>