Using Structure Tables with Python Operator in Generation 1 Graphs

Regarding your point in question 1), where one record being updated when 50 records are going in, I am not sure what you are referring there.  Are you saying that 50 records come in a single input or are you saying 50 records come in and it only handles one record at a time and not batching things up?  I would think you need to setup a batch logic in the python code if it is the latter.  Consider the following code to batch up 50 records.

counter=0
body =[]
def on_input(data):
   global body
   global counter
   inputdata = data.split(',')
   body.append(inputdata)
   type_id ="mytbl"
   counter = counter + 1
   if counter == 50:
     batch1 = api.Table(body, type_id, {"lastBatch": True})
     api.send("output", batch1)
     api.send("debug", "50")
     counter =0
     body = []


api.set_port_callback("input1", on_input)

If it is the former then your logic is not separating the rows in the python code.  The batch size setting would not have any affect in this regard, you would need to add batching logic.

Regarding your third question, I do not understand or have the context to answer.

With respect to question 1), I am not familiar with the Kafka operators to that level of depth.  I suggest you open a support ticket for this type of question.

Regarding your question on 3) I have no view into the logic you are using to put the data together for insertion into the database.  I don’t think the Table Producer is changing your data if that is what you are asking.  I would consider two things to debug your logic:

a. Perhaps send the data to a debug port to validate what data looks like at certain times within your script.  You like would need to “stringify” the data and send it to a wiretap.  (ie api.send("debugport",str(mytbl)) )

b. Truncate the table so that you are not looking at older data and older runs which may have had a problem that was fixed.

Regarding your question 4), you can setup a parameter in your python operator by pressing the "+" button in the operators configuration boxed in red in the image below.

Then provide the parameter a name.  As shown below, I will call it "getsize".

Notice the value "${sizeprompt}"  I provide above. I will reference this later when I run the graph.

From this point we have a configuration parameter for the python operator that can be used in your python code as follows:

batchsize = api.config.getsize

This will set the value of your python variable to the setting used in the python configuration.

Save the Graph and Run it.

You will see a prompt for "sizeprompt" when running the graph because I used ${} around the value this markup treats the python parameter as a runtime parameter for the graph and the number you specify in the run setting will be used in your python code to set the value of the batch size.

I hope this helps.

Regarding your question 2) you do not have a variable called "mytable" in your code and you are referencing an undefined variable.  I used this variable("mytbl") as it was referenced in previous posts.   In the code you provided, you would most likely be interested in "batch1"  so you would want to send batch1 to the "debugport".  At least that is how I interpret it.

api.send("debugport",str(batch1))

Regarding question 3, this is more of a general Python question.  In my blog example, the data coming into the Python operator is CSV so splitting via comma was what was used.  If the data is JSON or string representation of JSON, then you need to treat the data accordingly.

First it would be easiest to handle the data as a JSON object using the following code.

##if the message is string data
if type(data) is str:
  data = json.loads(data)

I am not certain about the entire message as it could have multiple rows or a single row entry per message.  So you as a reader have to apply your understanding of the data.

In the simplest case where there is only one record per message, then you could provide the following code to format the data assuming it is a JSON object.

##if there is one record per message
item = data["scheduleEntries"]["scheduleEntry"][0]
inputdata = [item["rotationId"],item["rotationName"],item["rotationVersion"],item["scheduleEntryID"]["scheduleEntryKey"]]

If there are multiple rows, then this is more complicated as you will want to loop through the object with something similar to the following:

datain = data["scheduleEntries"]["scheduleEntry"]
for item in datain:
  inputdata = [item["rotationId"],item["rotationName"],item["rotationVersion"],item["scheduleEntryID"]["scheduleEntryKey"]]
  body.append(inputdata)

Since I am only guessing at the your data structure and desired output, this code may need to be changed for the actual format and desired output.

There are three problems with the code as I understand your intentions.

First, your indentations are not consistent and this is the reason for the errors above.  You started with a three character indentation within your function and when you have your first if statement and "for loop" you have a seven character indentation.  With Python indents matter.

Second, your "datain" variable may not be set if the body variable is a JSON object and not a string as this variable is only set when it incoming data is a string.  Your current indentation has the "datain" assignment set when the data variable is a string.

The third problem is the data and your code are not aligned.  Your looping does not take into account the arrays in your data.  Also the dictionary references are also not likely going to work once your indentations are array references are fixed.

Consider the following code corrections:

counter=0
body =[]
def on_input(data):
   global body
   global counter
   ##if the message is string data
   ##Correction for indentations
   if type(data) is str:
      data = json.loads(data)
   ## Correction for nested lists
   ##datain = data["scheduleEntries"]["scheduleEntry"]
   for row in datain:
      ##Correction on field references
      ##inputdata = [item["rotationId"],item["rotationName"],item["rotationVersion"],item["scheduleEntryID"]["scheduleEntryKey"],item["scheduleEntryIdentifier"]["previousPortCall"]["terminalCode"]["actualArrival"]]
      ##Assumption that only one record/first record in the second array.
      item=row[0]["scheduleEntries"]["scheduleEntry"][0]
      inputdata = [item["rotationId"],item["rotationName"],item["rotationVersion"],item["scheduleEntryID"]["scheduleEntryKey"],item["scheduleEntryID"]["scheduleEntryIdentifier"]["previousPortCall"]["terminalCode"],item["scheduleEntryID"]["scheduleEntryIdentifier"]["previousPortCall"]["actualArrival"]]      
      body.append(inputdata)
   type_id ="mytable"
   counter = counter + 1
   if counter == 50:
      batch1 = api.Table(body, type_id, {"lastBatch": True})
      api.send("output", batch1)
      api.send("output", "50")
      api.send("debugport",str(batch1))
      counter =0
      body = []


api.set_port_callback("input1", on_input)

Again the "standard disclaimer", this code is subject to my interpretation of the desired result and data structures and may not completely accurate.

As this seems to be more of a general Python set of question, there are many Python resources on the web for general Python learning you may want to have a look at the following sites for general Python questions like these.

OpenSAP

W3 Schools