This demo showcases how to automate and monitor Airbyte Cloud sync jobs using PyAirbyte. It includes setting up job executions, handling dependencies, sending real-time status updates, and visually representing job details and outcomes on a timeline.
!apt-get install -qq python3.10 -venv
%pip install --quiet airbyteJob execution and notifications using PyAirbyte Here, we initialize an Airbyte Cloud workspace with PyAirbyte to manage data sync jobs. We show how Slack can be used for real-time notifications on job statuses.
The code below does the following:
Workspace initialization: Sets up the workspace with necessary credentials. Slack notifications: Configures a webhook for sending job statuses to a Slack channel. Job data tracking: Collects and stores execution details like start time, duration, and status for visualization. Run syncs: Executes jobs sequentially based on dependency success and sends detailed outcomes to Slack (or simply print them). import requests
from airbyte import cloud
from datetime import datetime
from google.colab import userdata
workspace = cloud.CloudWorkspace(
workspace_id=userdata.get("AIRBYTE_WORKSPACE_ID" ),
api_key=userdata.get("AIRBYTE_API_KEY" ),
)
slack_webhook_url = "https://hooks.slack.com/services/XXX/YYY/ZZZ"
job_data = []
def send_slack_message (message, webhook_url ):
"""Sends a message to a Slack channel specified by the webhook URL."""
payload = {'text' : message}
response = requests.post(webhook_url, json=payload)
return response.status_code
def trigger_sync_and_check (connection_id ):
"""Triggers a sync job for a specified connection ID and checks the job status."""
start_time = datetime.now()
connection = workspace.get_connection(connection_id=connection_id)
sync_result = connection.run_sync()
end_time = datetime.now()
duration = (end_time - start_time).total_seconds()
status = str (sync_result.get_job_status())
job_data.append({
'connection_id' : connection_id,
'start_time' : start_time,
'end_time' : end_time,
'duration' : duration,
'status' : status
})
return status, sync_result, start_time, duration
def format_sync_details (sync_result, start_time, duration ):
"""Format sync details for Slack message, including start time and duration."""
details = (
f"Job ID: {sync_result.job_id} \n"
f"Start Time: {start_time.strftime('%Y-%m-%d %H:%M:%S' )} \n"
f"Duration: {duration:.2 f} seconds\n"
f"Records Synced: {sync_result.records_synced} \n"
f"Bytes Synced: {sync_result.bytes_synced} \n"
f"Job URL: {sync_result.job_url} "
)
return details
def run_syncs ():
"""Handle sequential sync jobs where the second job depends on the success of the first."""
first_connection_id = "df2898ce-b21f-43c0-8c82-bd8c6dd24931"
second_connection_id = "86075252-9004-41fc-b5e3-f61f20388ff5"
first_status, first_sync_result, first_start, first_duration = trigger_sync_and_check(first_connection_id)
if "SUCCEEDED" in first_status:
message = f"First sync succeeded (Connection ID: {first_connection_id} ).\nDetails:\n{format_sync_details(first_sync_result, first_start, first_duration)} \nProceeding to second sync."
send_slack_message(message, slack_webhook_url)
print (message)
second_status, second_sync_result, second_start, second_duration = trigger_sync_and_check(second_connection_id)
if "SUCCEEDED" in second_status:
message = f"Second sync succeeded (Connection ID: {second_connection_id} ).\nDetails:\n{format_sync_details(second_sync_result, second_start, second_duration)} "
else :
message = f"Second sync failed (Connection ID: {second_connection_id} ). Status: {second_status} \nDetails:\n{format_sync_details(second_sync_result, second_start, second_duration)} "
else :
message = f"First sync failed (Connection ID: {first_connection_id} ). Status: {first_status} \nDetails:\n{format_sync_details(first_sync_result, first_start, first_duration)} \nSecond sync not initiated."
send_slack_message(message, slack_webhook_url)
print (message)First sync succeeded (Connection ID: df2898ce-b21f-43c0-8c82-bd8c6dd24931).
Job execution timeline visualization With the data collected from the syncs we previously ran using PyAirbyte, we can create a visualization that provides a clear and intuitive representation of a series of job executions over time.
Each job is displayed as a rectangle along a timeline, where the width of each rectangle corresponds to the job's duration. The colors indicate the status of each job—green for successful completions and yellow for failures. This graph is particularly useful for quickly assessing the overall health and efficiency of a workflow. It allows viewers to see at a glance:
Job Durations: How long each job took to complete. Job Start Times: When each job began relative to the start of the timeline. Job Outcomes: The success or failure of each job. import matplotlib.pyplot as plt
import matplotlib.patches as patches
from datetime import datetime, timedelta
def get_color (status ):
"""Returns color based on the job status."""
if "SUCCEEDED" in status:
return 'green'
elif "FAILED" in status:
return 'yellow'
else :
return 'gray'
def draw_job_graph (job_data ):
"""Draws a graph of job executions where rectangles represent the duration and color indicates status."""
fig, ax = plt.subplots(figsize=(14 , 3 ))
start_time = min (job['start_time' ] for job in job_data)
end_time = max (job['start_time' ] + timedelta(seconds=job['duration' ]) for job in job_data)
total_duration = (end_time - start_time).total_seconds()
for job in job_data:
start_offset = (job['start_time' ] - start_time).total_seconds()
width = job['duration' ]
color = get_color(job['status' ])
rect = patches.Rectangle((start_offset, 0 ), width, 1 , linewidth=1 , edgecolor='black' , facecolor=color)
ax.add_patch(rect)
ax.text(start_offset + width / 2 , 0.5 , f"{job['connection_id' ]} ({job['duration' ]} s)" ,
verticalalignment='center' , horizontalalignment='center' , color='black' , fontweight='bold' )
ax.set_xlim(0 , total_duration)
ax.set_ylim(-0.5 , 1.5 )
ax.set_xlabel("Time (seconds from start)" )
ax.set_title('Job Execution Timeline' )
plt.xticks(range (0 , int (total_duration) + 1 , 20 ))
plt.yticks([])
plt.grid(True , which='both' , axis='x' , linestyle='--' , linewidth=0.5 )
plt.show()Visualize sync results
Airbyte