code: @Override
      public void execute(TridentTuple tuple, TridentCollector collector)
      {
          String receivedStr = tuple.getString(0);
  
          String[] splitedStr = StringUtils.split(receivedStr, this.delimeter);
          int dataNum = splitedStr.length;
          double[] points = new double[splitedStr.length];
  
          try
          {
              for (int index = 0; index < dataNum; index++)
              {
                  points[index] = Double.parseDouble(splitedStr[index].trim());
              }
  
              KmeansPoint result = new KmeansPoint();
              result.setDataPoint(points);
              collector.emit(new Values(result));
          }
          catch (Exception ex)
          {
              logger.warn("Received data is invalid. skip this data. ReceivedData=" + receivedStr, ex);
          }
      }
docs: {@inheritDoc}