I would like to be able to separate a feature into smaller chunks and process them 1 by 1
I have managed to use ModuloCounter to separate the feature but I am unsure how to get the rest of the workspace to process the broken up features 1 by 1.
If I send all the output into pythonCaller they are executed together at the same time.
Any ideas?
+28a) you should be able to alter your python code to process the features individually -- posting the code, or at least the pseudo structure would help.
b) if you want 1 by 1 you're better off with a counter rather than a modulo counter
c) you can wrap the pythonCaller in a custom transformer and set the parallel process group by to your unique ID attribute (with or without actual parallel processing)
Also what is the AttributeRangeFilter doing?
a) you should be able to alter your python code to process the features individually -- posting the code, or at least the pseudo structure would help.
b) if you want 1 by 1 you're better off with a counter rather than a modulo counter
c) you can wrap the pythonCaller in a custom transformer and set the parallel process group by to your unique ID attribute (with or without actual parallel processing)
Also what is the AttributeRangeFilter doing?
+7Hi @ashertbrooks,
It is not quite clear what exactly you want to do. Is it possible to describe the problem a bit more in depth and provide the Python code? By default a PythonCaller does process features one by one. Making sure that features enter the PythonCaller via different connection lines does not change anything to that.
+28
Also a bit of terminology that might help clarify responses. You have one dataset (the csv), you have 38070 features, each one is a row in the csv. The arrows on each transformer are called ports, The lines between each port are connections.
Looking at the screenshot the first feature (1st data row in the csv) get's assigned a value of 0 by the modulo counter, two attribute manipulations are performed on the feature (string replacers), it is routed by the AttributeRangeFilter (ARF) through the first port which is connected to the pythonCaller. the second feature has a value of 1 in the modulo counter, has the same manipulations, is filtered through the second port on the ARF, but ends up at the same pythonCaller.
Hi @ashertbrooks,
It is not quite clear what exactly you want to do. Is it possible to describe the problem a bit more in depth and provide the Python code? By default a PythonCaller does process features one by one. Making sure that features enter the PythonCaller via different connection lines does not change anything to that.
By breaking the 50,000 records into 10 chunks of 5,000 and process them one after the other, it is much faster and if it errors due to data formatting than I have only lost 10% of the data rather than the whole lot failing.
Also a bit of terminology that might help clarify responses. You have one dataset (the csv), you have 38070 features, each one is a row in the csv. The arrows on each transformer are called ports, The lines between each port are connections.
Looking at the screenshot the first feature (1st data row in the csv) get's assigned a value of 0 by the modulo counter, two attribute manipulations are performed on the feature (string replacers), it is routed by the AttributeRangeFilter (ARF) through the first port which is connected to the pythonCaller. the second feature has a value of 1 in the modulo counter, has the same manipulations, is filtered through the second port on the ARF, but ends up at the same pythonCaller.
Hi @ashertbrooks,
It is not quite clear what exactly you want to do. Is it possible to describe the problem a bit more in depth and provide the Python code? By default a PythonCaller does process features one by one. Making sure that features enter the PythonCaller via different connection lines does not change anything to that.
tmp = '{"REGISTER_NAME" : "' + feature.getAttribute('REGISTER_NAME') + '", "ADV_NAME" : "' + feature.getAttribute("ADV_NAME") + '", and so on... + '"}'
feature.setAttribute("JSON_DATA", tmp) #json.dumps(tmp)
What I would suggest is using the JSON templater.
+7Thanks
What the code does is count how many features have entered already and when this count exceeds a certain number (5 in my case), it will export a feature containing the json.
import fme
import fmeobjects
class FeatureProcessor(object):
# Initialise some variables before the first feature enters
def __init__(self):
self.nbFeaturesProcessed = 0
self.json = "{"
# This code is executed everytime a feature enters the PythonCaller
def input(self,feature):
self.json += "'CREATION_INSTANCE':{},".format(feature.getAttribute('_creation_instance'))
self.nbFeaturesProcessed += 1
# Check if already 5 features are processed
# P.S. You can make a parameter from the number 10 0000 ;)
if self.nbFeaturesProcessed >= 5:
self.export()
# This method is called after the last feature has entered
def close(self):
# If some features where added but the amount is < 5,
# we should make sure those are exported also
if self.nbFeaturesProcessed > 0:
self.export()
# Method that is called to export the created json
def export(self):
# Finalise the json-file by removing the last comma and adding the
# curly bracket at the end
json = "{}{}".format(self.json[:-1], '}')
# Create a new feature having only one attribute containing the json
newFeature = fmeobjects.FMEFeature()
newFeature.setAttribute('json', json)
self.pyoutput(newFeature)
# reset the variables to start over again
self.nbFeaturesProcessed = 0
self.json = "{"I just create 10 features in the creator and as you can see, those are combined into 2 groups containing 5 features each.
If you still have questions. Please ask!
+7Hi @ashertbrooks,
It is not quite clear what exactly you want to do. Is it possible to describe the problem a bit more in depth and provide the Python code? By default a PythonCaller does process features one by one. Making sure that features enter the PythonCaller via different connection lines does not change anything to that.
Thanks
What the code does is count how many features have entered already and when this count exceeds a certain number (5 in my case), it will export a feature containing the json.
import fme
import fmeobjects
class FeatureProcessor(object):
# Initialise some variables before the first feature enters
def __init__(self):
self.nbFeaturesProcessed = 0
self.json = "{"
# This code is executed everytime a feature enters the PythonCaller
def input(self,feature):
self.json += "'CREATION_INSTANCE':{},".format(feature.getAttribute('_creation_instance'))
self.nbFeaturesProcessed += 1
# Check if already 5 features are processed
# P.S. You can make a parameter from the number 10 0000 ;)
if self.nbFeaturesProcessed >= 5:
self.export()
# This method is called after the last feature has entered
def close(self):
# If some features where added but the amount is < 5,
# we should make sure those are exported also
if self.nbFeaturesProcessed > 0:
self.export()
# Method that is called to export the created json
def export(self):
# Finalise the json-file by removing the last comma and adding the
# curly bracket at the end
json = "{}{}".format(self.json[:-1], '}')
# Create a new feature having only one attribute containing the json
newFeature = fmeobjects.FMEFeature()
newFeature.setAttribute('json', json)
self.pyoutput(newFeature)
# reset the variables to start over again
self.nbFeaturesProcessed = 0
self.json = "{"I just create 10 features in the creator and as you can see, those are combined into 2 groups containing 5 features each.
If you still have questions. Please ask!
Thanks jeroenstiers
Thanks
What the code does is count how many features have entered already and when this count exceeds a certain number (5 in my case), it will export a feature containing the json.
import fme
import fmeobjects
class FeatureProcessor(object):
# Initialise some variables before the first feature enters
def __init__(self):
self.nbFeaturesProcessed = 0
self.json = "{"
# This code is executed everytime a feature enters the PythonCaller
def input(self,feature):
self.json += "'CREATION_INSTANCE':{},".format(feature.getAttribute('_creation_instance'))
self.nbFeaturesProcessed += 1
# Check if already 5 features are processed
# P.S. You can make a parameter from the number 10 0000 ;)
if self.nbFeaturesProcessed >= 5:
self.export()
# This method is called after the last feature has entered
def close(self):
# If some features where added but the amount is < 5,
# we should make sure those are exported also
if self.nbFeaturesProcessed > 0:
self.export()
# Method that is called to export the created json
def export(self):
# Finalise the json-file by removing the last comma and adding the
# curly bracket at the end
json = "{}{}".format(self.json[:-1], '}')
# Create a new feature having only one attribute containing the json
newFeature = fmeobjects.FMEFeature()
newFeature.setAttribute('json', json)
self.pyoutput(newFeature)
# reset the variables to start over again
self.nbFeaturesProcessed = 0
self.json = "{"I just create 10 features in the creator and as you can see, those are combined into 2 groups containing 5 features each.
If you still have questions. Please ask!
Thanks again for the help
+7Thanks
What the code does is count how many features have entered already and when this count exceeds a certain number (5 in my case), it will export a feature containing the json.
import fme
import fmeobjects
class FeatureProcessor(object):
# Initialise some variables before the first feature enters
def __init__(self):
self.nbFeaturesProcessed = 0
self.json = "{"
# This code is executed everytime a feature enters the PythonCaller
def input(self,feature):
self.json += "'CREATION_INSTANCE':{},".format(feature.getAttribute('_creation_instance'))
self.nbFeaturesProcessed += 1
# Check if already 5 features are processed
# P.S. You can make a parameter from the number 10 0000 ;)
if self.nbFeaturesProcessed >= 5:
self.export()
# This method is called after the last feature has entered
def close(self):
# If some features where added but the amount is < 5,
# we should make sure those are exported also
if self.nbFeaturesProcessed > 0:
self.export()
# Method that is called to export the created json
def export(self):
# Finalise the json-file by removing the last comma and adding the
# curly bracket at the end
json = "{}{}".format(self.json[:-1], '}')
# Create a new feature having only one attribute containing the json
newFeature = fmeobjects.FMEFeature()
newFeature.setAttribute('json', json)
self.pyoutput(newFeature)
# reset the variables to start over again
self.nbFeaturesProcessed = 0
self.json = "{"I just create 10 features in the creator and as you can see, those are combined into 2 groups containing 5 features each.
If you still have questions. Please ask!
I am glad I could help!
Hi @ashertbrooks, if you want to make groups for every N features in the input order, you can add group ID attribute which can be computed with this expression to each feature.
@int(@Count() / N)
Here, N indicates the number of features per group. You can also use the Counter transformer instead of the @Count function.
Then, this workflow should work as well as the Python script provided by @jeroenstiers.
Hope this helps.
+7Hi @ashertbrooks, if you want to make groups for every N features in the input order, you can add group ID attribute which can be computed with this expression to each feature.
@int(@Count() / N)
Here, N indicates the number of features per group. You can also use the Counter transformer instead of the @Count function.
Then, this workflow should work as well as the Python script provided by @jeroenstiers.
Hope this helps.
Hi @ashertbrooks, if you want to make groups for every N features in the input order, you can add group ID attribute which can be computed with this expression to each feature.
@int(@Count() / N)
Here, N indicates the number of features per group. You can also use the Counter transformer instead of the @Count function.
Then, this workflow should work as well as the Python script provided by @jeroenstiers.
Hope this helps.
Enter your E-mail address. We'll send you an e-mail with instructions to reset your password.
