Merge branch 'master' of https://github.com/ertis-research/kafka-ml

federated-module/federated_model_training/tensorflow/README.md

@@ -7,7 +7,7 @@ A brief introduction of its files:
 - File `KafkaModelEngine.py` class contains all the necesary functions to read and load Tensorflow models from Apache Kafka in cloud.
 - File `FederatedKafkaMLAggregationSink.py` class contains all the necesary functions to send the trained model and metrics to Apache Kafka in cloud.
 - File `federated_mainTraining.py` and `federated_singleClassicTraining.py` contains the functions for each kind of federated training.
-- File `decoders.py` decoders (RAW, Avro, JSON) used to decode data streams.
+- File `decoders.py` decoders (RAW, Avro, JSON, TELEGRAF_STR_JSON) used to decode data streams.
 - File `config.py` to configure debug.
 - File `utils.py` common functions used by other files.
 

federated-module/federated_model_training/tensorflow/decoders.py

@@ -70,12 +70,13 @@ def decode(self, x, y):
         return (res_x, res_y)
 
 class JsonDecoder:
-    """JSON class decoder implementation
-        ARGS:
-            configuration (dic): configuration properties
-        Attributes:
-            scheme(str): scheme of the JSON implementation
+    """JSON class decoder implementation"""
 
-    """
     def decode(self, x):
-        return json.loads(x)
+        return json.loads(x)
+
+class TelegrafStringJsonDecoder:
+    """TELEGRAF_STR_JSON class decoder implementation"""
+
+    def decode(self, x):
+        return json.loads(json.loads(x)["fields"]["value"])

model_inference/pytorch/decoders.py

@@ -69,12 +69,13 @@ def decode(self, x, y):
         return res
 
 class JsonDecoder:
-    """JSON class decoder implementation
-        ARGS:
-            configuration (dic): configuration properties
-        Attributes:
-            scheme(str): scheme of the JSON implementation
+    """JSON class decoder implementation"""
 
-    """
     def decode(self, x):
-        return json.loads(x)
+        return json.loads(x)
+
+class TelegrafStringJsonDecoder:
+    """TELEGRAF_STR_JSON class decoder implementation"""
+
+    def decode(self, x):
+        return json.loads(json.loads(x)["fields"]["value"])

model_inference/tensorflow/README.md

@@ -4,7 +4,7 @@ This module contains the inference task that will be executed when a TensorFlow
 
 A brief introduction of its files:
 - File `inference.py` main file of this module that will be executed when executed the inference Job.
-- File `decoders.py` decoders (RAW, Avro) used to decode data streams.
+- File `decoders.py` decoders (RAW, Avro, JSON, TELEGRAF_STR_JSON) used to decode data streams.
 - File `config.py` to configure debug.
 - File `utils.py` common functions used by other files.
 

model_inference/tensorflow/decoders.py

@@ -15,6 +15,8 @@ def get_decoder(input_format, configuration):
             return AvroDecoder(configuration)
         elif input_format == 'JSON':
             return JsonDecoder()
+        elif input_format == 'TELEGRAF_STR_JSON':
+            return TelegrafStringJsonDecoder()
         else:
             raise ValueError(input_format)
 
@@ -57,12 +59,13 @@ def decode(self, msg):
         return res
 
 class JsonDecoder:
-    """JSON class decoder implementation
-        ARGS:
-            configuration (dic): configuration properties
-        Attributes:
-            scheme(str): scheme of the JSON implementation
+    """JSON class decoder implementation"""
 
-    """
     def decode(self, x):
-        return json.loads(x)
+        return json.loads(x)
+
+class TelegrafStringJsonDecoder:
+    """TELEGRAF_STR_JSON class decoder implementation"""
+
+    def decode(self, x):
+        return json.loads(json.loads(x)["fields"]["value"])

model_training/tensorflow/README.md

@@ -4,7 +4,7 @@ This module contains the training task that will be executed when a TensorFlow t
 
 A brief introduction of its files:
 - File `training.py` main file of this module that will be executed when executed the training Job.
-- File `decoders.py` decoders (RAW, Avro) used to decode data streams.
+- File `decoders.py` decoders (RAW, Avro, JSON, TELEGRAF_STR_JSON) used to decode data streams.
 - File `config.py` to configure debug.
 - File `utils.py` common functions used by other files.
 

model_training/tensorflow/decoders.py

@@ -70,12 +70,13 @@ def decode(self, x, y):
         return (res_x, res_y)
 
 class JsonDecoder:
-    """JSON class decoder implementation
-        ARGS:
-            configuration (dic): configuration properties
-        Attributes:
-            scheme(str): scheme of the JSON implementation
+    """JSON class decoder implementation"""
 
-    """
     def decode(self, x):
-        return json.loads(x)
+        return json.loads(x)
+
+class TelegrafStringJsonDecoder:
+    """TELEGRAF_STR_JSON class decoder implementation"""
+
+    def decode(self, x):
+        return json.loads(json.loads(x)["fields"]["value"])