Conclusions
5.5 Combining work and care
Es el encargado de procesar todos los datos en tiempo real, añadiendo agregaciones que serán útil tanto para la administración del sistema como de la medición de la calidad del servicio.
Pipes FillContainerIn o Float: fill o Date: timeStamp o Integer: id o Double: lat o Double: lng FillContainerInString o String: JSONString FillContainerInConverted o String: fill o String: timeStamp o String: id o String: lat o String: lng FillContainerDayIn
o Array Float: fill
o Date: timeStamp
o Integer: id
FillContainerDayPred
o Array Float: fill
o Integer: id
FillContainerDayPredConverted
o Array Float: fill
o Date: timeStamp o Integer: id o Integer: index FillContainerDayQuarts o Float: firstQuart o Float: secondQuart o Float: thirdQuart o Float: fourthQuart o Date: timeStamp o Integer: id FillContainerDayQuartsConverted o String: firstQuart o String: secondQuart o String: thirdQuart o String: fourthQuart o String: timeStamp o String: id FillContainerMinMaxMean o Float: min o Float: max o Float: mean o Date: timeStamp
o Integer: id FillContainerMinMaxMeanConverted o Float: min o Float: max o Float: mean o Date: timeStamp o Integer: id Filters
El parámetro “partition by” significa que trabaja de manera completamente independiente con las tuplas que tengan diferentes IDs, es decir que, si hace una media o un sumatorio de 20 tuplas, lo hace por separado para cada ID, hace la media o el sumatorio de las 20 tuplas que tengan el mismo ID.
En el caso de los componentes que escriban sobre repositorios, en los parámetros de output cuando están entre “comillas” significa que el valor siempre es ese tupla tras tupla, si no está entre comillas es un valor tomado de la tupla que ha llegado.
InputSource:
o Output: FillContainerInString
KafkaImporter:
o Output: FillContainerInString
Lee las tuplas que le llegan por kafka y las envía por la salida. Usa el toolkit com.ibm.streamsx.messaging.
JSONToTuple
o Input: FillContainerInString
o OutPut: FillContainerIn
Usa el toolkit com.ibm.streamsx.json
UnionInput:
o Input: FillContainerIn
o Output: FillContainerIn
Todo lo que le llega por las entradas los emite por las salidas.
Diezmator:
o Input: FillContainerIn
o Output: FillContainerIn
o Partition by: id
La versión de producción lo que hace es una media de las tuplas que han llegado en los últimos 10 minutos, en la versión para el simulador por cada 20 tuplas que le llegan, envía una cuyo valor de llenado es la media de las 20 recibidas y la etiqueta de tiempo de la última recibida.
AllDayPack:
o Input: FillContainerIn
o Output: FillContainerDayIn
Cuando recibe todas las tuplas de un día (72) envía una tupla por la sálica con la fecha de la última recibida, el id de la partición y un array con todas las tuplas del día juntas.
CountDay:
o Input: FillContainerDayIn.
o Output: FillContainerDayQuarts
Por cada tupla que le llega haya sobre el array de llenados el porcentaje de ellas que se encuentran entre [<0.25, 0.25-0.50, 0.50-0.75, >0.75], y los envía por las salidas junto al id y a la etiqueta de tiempo recibida.
ConversorCountDay:
o Input: FillContainerDayQuarts
o Output: FillContainerDayQuartsConverted
Por cada tupla que le llega convierte los campos a String para poder introducirlo en el repositorio y poder leer los datos directamente.
SinkCountFirst:
o Input: FillContainerDayQuartsConverted
o Output:
Table:” countDay”
Row Key: timeStamp
Column family:” firstQuart”
Column qualifiers: id
Value: firstQuart
Por cada tupla que le llega convierte la introduce en el repositorio Hbase siguiendo los campos descritos en output.
Hace uso del toolkit com.ibm.streamsx.hbase 2.0.0.
o Input: FillContainerDayQuartsConverted
o Output:
Table:” countDay”
Row Key: timeStamp
Column family: “secondQuart”
Column qualifiers: id
Value: secondQuart
Por cada tupla que le llega convierte la introduce en el repositorio Hbase siguiendo los campos descritos en output.
Hace uso del toolkit com.ibm.streamsx.hbase 2.0.0.
SinkCountThird:
o Input: FillContainerDayQuartsConverted
o Output:
Table: “countDay”
Row Key: timeStamp
Column family: “thirdQuart”
Column qualifiers: id
Value: thirdQuart
Por cada tupla que le llega convierte la introduce en el repositorio Hbase siguiendo los campos descritos en output.
Hace uso del toolkit com.ibm.streamsx.hbase 2.0.0.
SinkCountFourth:
o Input: FillContainerDayQuartsConverted
o Output:
Table: “countDay”
Column family: “fourthQuart”
Column qualifiers: id
Value: fourthQuart
Por cada tupla que le llega convierte la introduce en el repositorio Hbase siguiendo los campos descritos en output.
Hace uso del toolkit com.ibm.streamsx.hbase 2.0.0.
AutoFore:
o Input: FillContainerIn.
o Output: FillContainerDayPred
o Partition by: id
Por cada tupla que le llega hace una predicción de los valores de llenado de los 2 días siguientes (144 mediciones) en la salida introduce un array con los 144 valores, otro con las 144 etiquetas de tiempo y el Id del contenedor.
El método utilizado para la predicción es el uso de las mediciones de la última semana como mínimo (504 mediciones) junto con métodos de análisis como son regresión lineal, ARIMA etc.
Hace uso del toolkit com.ibm.streams.timeseries.
ConversorCountDay:
o Input: FillContainerDayPred
o Output: FillContainerInConverted
Por cada tupla que le llega, recorre todos los valores de llenados y de etiquetas de tiempo y los envía por separado por la salida convirtiendo los campos a String añadiéndole un índice para poder introducirlo en el repositorio y poder leer los datos directamente.
o Input: FillContainerDayPredConverted
o Output:
Table: “predictions”
Row Key: id
Column family:” values”
Column qualifiers: index
Value: timeStamp
Por cada tupla que le llega convierte la introduce en el repositorio Hbase siguiendo los campos descritos en output, en este caso sobre escribe la tupla anterior con mismo id y con mismo índice.
Hace uso del toolkit com.ibm.streamsx.hbase 2.0.0.
SinkPredValue:
o Input: FillContainerDayPredConverted
o Output:
Table: “predictions”
Row Key: id
Column family: “values”
Column qualifiers: index
Value: fill
Por cada tupla que le llega convierte la introduce en el repositorio Hbase siguiendo los campos descritos en output, en este caso sobre escribe la tupla anterior con mismo id y con mismo índice.
Hace uso del toolkit com.ibm.streamsx.hbase 2.0.0.
ConversorMeasures:
o Input: FillContainerIn
Por cada tupla que le llega convierte los campos a String para poder introducirlo en el repositorio y poder leer los datos directamente.
SinkMeasures:
o Input: FillContainerInConverted
o Output:
Table: “measures”
Row Key: timeStamp
Column family: “container”
Column qualifiers: id
Value: fill
Por cada tupla que le llega convierte la introduce en el repositorio Hbase siguiendo los campos descritos en output.
Hace uso del toolkit com.ibm.streamsx.hbase 2.0.0.
SinkLastMEasure:
o Input: FillContainerInConverted
o Output:
Table: “lastMeasures”
Row Key: id
Column family:” lastFill”
Column qualifiers:” fill”
Value: fill
Por cada tupla que le llega convierte la introduce en el repositorio Hbase siguiendo los campos descritos en output, en este caso sobrescribe la tupla anterior con mismo id.
Integer:
o Input: FillContainerIn.
o Output: FillContainerIn
o Partition by: id
Por cada tupla que le llega hace el sumatorio de ella y las 72 anteriores (un día de mediciones) y la divide entre el número de mediciones sumadas, es decir hace la media de ella y de las 71 anteriores mediciones por cada tupla que entra. La salida es el resultado de esa media y la id y la fecha de la última que entro
ConversorInteger:
o Input: FillContainerIn
o Output: FillContainerInConverted
Por cada tupla que le llega, recorre todos los valores de llenados y de etiquetas de tiempo y los envía por separado por la salida convirtiendo los campos a String para poder introducirlo en el repositorio y poder leer los datos directamente.
SinkInteger:
o Input: FillContainerInConverted
o Output:
Table:” integral”
Row Key: timeStamp
Column family:” container”
Column qualifiers: id
Value: fill
Por cada tupla que le llega convierte la introduce en el repositorio Hbase siguiendo los campos descritos en output.
Hace uso del toolkit com.ibm.streamsx.hbase 2.0.0.
o Input: FillContainerIn.
o Output: FillContainerMinMaxMean
o Partition by: id
Por cada 1440 tuplas que le llegan (1 día de mediciones), envía una que incluye el máximo, el mínimo y la media de las 1440 l recibidas y la etiqueta de tiempo de la última recibida.
ConversorMinMaxMean:
o Input: FillContainerMinMaxMeanConverted
o Output: FillContainerMinMaxMeanConverted
Por cada tupla que le llega convierte los campos a String para poder introducirlo en el repositorio y poder leer los datos directamente.
SinkMean:
o Input: FillContainerMinMaxMeanConverted
o Output:
Table: “MaxMeanMin”
Row Key: timeStamp
Column family:” MMM”
Column qualifiers:” mean”
Value: mean
Por cada tupla que le llega convierte la introduce en el repositorio Hbase siguiendo los campos descritos en output.
Hace uso del toolkit com.ibm.streamsx.hbase 2.0.0.
SinkMax:
o Input: FillContainerMinMaxMeanConverted
o Output:
Row Key: timeStamp
Column family:” MMM”
Column qualifiers:” max”
Value: max
Por cada tupla que le llega convierte la introduce en el repositorio Hbase siguiendo los campos descritos en output.
Hace uso del toolkit com.ibm.streamsx.hbase 2.0.0.
SinkMin:
o Input: FillContainerMinMaxMeanConverted
o Output:
Table:” MaxMeanMin”
Row Key: timeStamp
Column family:” MMM”
Column qualifiers:” min”
Value: min
Por cada tupla que le llega convierte la introduce en el repositorio Hbase siguiendo los campos descritos en output.
Hace uso del toolkit com.ibm.streamsx.hbase 2.0.0.
AutoForeMax:
o Input: FillContainerMinMaxMean.
o Output: FillContainerDayPred
o Partition by: id
Por cada tupla que le llega hace una predicción de los valores de llenado máximos de los 2 días siguientes (2 mediciones)
En la salida introduce un array con los 2 valores, otro con las 2 etiquetas de tiempo y el Id del contenedor.
El método utilizado para la predicción es el uso de las mediciones máximas del último año como mínimo (365 mediciones) junto con métodos de análisis como son regresión lineal, ARIMA etc.
Hace uso del toolkit com.ibm.streams.timeseries.
ConversorCountDay:
o Input: FillContainerDayPred
o Output: FillContainerDayPredConverted
Por cada tupla que le llega, recorre todos los valores de llenados y de etiquetas de tiempo y los envía por separado por la salida convirtiendo los campos a String y añadiendo un numero de índice para poder introducirlo en el repositorio y poder leer los datos directamente.
SinkAFMaxDate:
o Input: FillContainerDayPredConverted
o Output:
Table:” predicionsMax”
Row Key: id
Column family:” timestamps”
Column qualifiers: index
Value: timeStamp
Por cada tupla que le llega convierte la introduce en el repositorio Hbase siguiendo los campos descritos en output, en este caso sobre escribe la tupla anterior con mismo id y con mismo índice.
Hace uso del toolkit com.ibm.streamsx.hbase 2.0.0.
SinkAFmaxValue:
o Input: FillContainerDayPredConverted
Table:” predicionsMax”
Row Key: id
Column family:” values”
Column qualifiers: index
Value: fill
Por cada tupla que le llega convierte la introduce en el repositorio Hbase siguiendo los campos descritos en output, en este caso sobre escribe la tupla anterior con mismo id y con mismo índice
Ilustración 69 Diagrama de flujo del procesador principal .