Integration with Amazon SQS

Setting up

When using a Matillion ETL as part of a larger process the best practice way to initiate an orchestration job is to use Simple Queue Service or SQS. To do this, first create up to 3 SQS Queues. After this configure the screen below via the SQS Entry on the Project menu.

  • Enable SQS - This turns on or off the sub-system that listens to SQS queues.
  • Note: This is global, not for a particular project.
  • Credentials - Choose the credentials that you will be using to talk to SQS queues. An IAM user or Role will need the AmazonSQSFullAccess policy if you want to have a success or failure queue.
  • Region - the region where the queue exists in SQS.
  • Listen Queue - The name of the queue to listen for messages. These messages are in a set format as shown below.
{
  "group":"<Exactly matches the project group where your job resides>",
  "project":"<Exactly matches the project name where your job resides>",
  "version":"<Exactly matches the version Name>",
  "environment":"<Exactly matches the environment Name>",
  "job":"<Exactly matches the orchestration Job Name>",
  "variables":  {
               "<Name 1>": "<Value 1>",
               "<Name 2>": "<Value 3>"
           }
}
 

Note: The Variables are passed to the orchestration job. Matching variable names must be declared in the project with default values set for each environment. If a variable is passed which is not defined in the project, an error is logged in Project -> Task History

  • Enable Success - Enable this if you wish to place a message on an SQS queue when your orchestration job has completed. Success Queue - The name of the success queue on which to place success messages.
  • Compress - When ticked the body of the message on the queue will be gzipped. Use this to avoid hitting SQS limits.
  • Enable Failure - Enable this if you wish to place a message on a SQS queue when your orchestration job has failed.
  • Failure Queue - The name of the failure queue on which to place failure messages.
  • Compress - When ticked the body of the message on the queue will be gzipped. Use this to avoid hitting SQS limits.

Note: It is possible to use the same queue for success and failure messages. It’s strongly recommended that you do not use the listen queue for success and failure. Clicking the Test button will ensure that the queues exist and can be read.

Returned messages

The messages that are returned on the success or fail queues are in the following format:

{  
   "type":"QUEUE_ORCHESTRATION",
   "groupName":"<the group name that was executed>",
   "projectName":"<the project name that was executed>",
   "versionName":"<the version name that was executed>",
   "environmentName":"<the version name that was executed>",
   "state":"SUCCESS|FAILED", 
   "enqueuedTime":<Time message placed on queue in unix epoc format>,
   "startTime":<Time orchestration job began in unix epoc format>,
   "endTime":<Time orchestration job ended in unix epoc format>,
   "message":<contains error messages where applicable>,
   "originator ID":"q_Queue",

   "tasks":[  <This is a list of tasks executed in the orchestration>
       {  
           "type":"VALIDATE_ORCHESTRATION",
           "jobName":"SimpleQueueJob",
           "componentName":"Start 0",
           "orchestrationJobName":"SimpleQueueJob",
           "orchestrationComponentName":"Start 0",
           "state":"SUCCESS",
           "rowCount":0,
           "startTime":1443526622364,
           "endTime":1443526622364,
           "message":""
       },
       {  
           "type":"VALIDATE_ORCHESTRATION",
           "jobName":"SimpleQueueJob",
           "componentName":"End Success 0",
           "orchestrationJobName":"SimpleQueueJob",
           "orchestrationComponentName":"End Success 0",
           "state":"SUCCESS",
           "rowCount":0,
           "startTime":1443526622365,
           "endTime":1443526622369,
           "message":""
       },
       {  
           "type":"EXECUTE_ORCHESTRATION",
           "jobName":"SimpleQueueJob",
           "componentName":"End Success 0",
           "orchestrationJobName":"SimpleQueueJob",
           "orchestrationComponentName":"End Success 0",
           "state":"SUCCESS",
           "rowCount":0,
           "startTime":1443526622369,
           "endTime":1443526622369,
           "message":""
       }
   ],
   "rowCount":0
}

To put messages onto and SQS queue you can adapt this python snippet, or use any other AWS api at your disposal to achieve the same result.

import boto.sqs
import json
from boto.sqs.message import RawMessage

#connect to your region
conn = boto.sqs.connect_to_region("eu-west-1")

#create the queue to post messages
queue = conn.create_queue("QUEUE-NAME")

#set the queue to allow raw messages data
queue.set_message_class(RawMessage)

#prepare the message
p_message = {
 "created":"-",
 "group":"myGroup",
 "project":"myproject",
 "version":"default",
 "environment":"myEnv",
 "job":"myjob"
}

#prepare the message on the queue
message = queue.new_message(json.dumps(p_message))

#write the message to the queue
queue.write(message)