'Feeds Spark data partitions into the TensorFlow worker nodes.'

def train(self, dataRDD, num_epochs=0, qname='input'):

    logging.info('Feeding    training    data')
    assert (self.input_mode == InputMode.SPARK)
    assert (qname in self.queues)
    assert (num_epochs >= 0)
    if isinstance(dataRDD, DStream):
        dataRDD.foreachRDD((lambda rdd: rdd.foreachPartition(TFSparkNode.train(self.cluster_info, self.cluster_meta, qname))))
    else:
        if (num_epochs == 0):
            num_epochs = 10
        rdds = []
        for i in range(num_epochs):
            rdds.append(dataRDD)
        unionRDD = self.sc.union(rdds)
        unionRDD.foreachPartition(TFSparkNode.train(self.cluster_info, self.cluster_meta, qname))
