Сохранить изображение из системы RDD Apache Spark

Question

Я хочу получить изображения, которые я сохранил в моей системе RDD после их сопоставления.

Я создал простую сессию Spark в моем файле main.py, которая вызывает функцию preprocess_spark, которая возвращает массивкортежи с именем samples.Эти кортежи в формате (slide_num, image).Изображение - это np.array, которое будет преобразовано в изображение с помощью функции save_jpeg_help.

Когда я открываю веб-интерфейс Apache Spark, я вижу, что у него есть задание, соответствующее строке:

rdd.foreach(lambda sample_element: save_nonlabelled_sample_2_jpeg(sample_element, save_dir))

, но когда оно заканчивается, в моем каталоге save_dir ничего не сохраняется.

Есть идеи, что я делаю не так?

С уважением

main.py

spark = (SparkSession.builder
     .appName("Oncofinder -- Preprocessing")
     .getOrCreate())

samples = preprocess_spark(spark, [1])

if save_jpegs: #SET TO TRUE
    save_rdd_2_jpeg(samples, './data/images')


def save_rdd_2_jpeg(rdd, save_dir):
    rdd.foreach(lambda sample_element: save_nonlabelled_sample_2_jpeg(sample_element, save_dir))


def save_nonlabelled_sample_2_jpeg(sample, save_dir):
    slide_num, img_value = sample
    filename = '{slide_num}_{hash}.jpeg'.format(
        slide_num=slide_num, hash=np.random.randint(1e4))
    filepath = os.path.join(save_dir, filename)
    save_jpeg_help(img_value, filepath)

def save_jpeg_help(img_value, filepath):
    dir = os.path.dirname(filepath)
    os.makedirs(dir, exist_ok=True)
    img = Image.fromarray(img_value.astype(np.uint8), 'RGB')
    img.save(filepath)


def preprocess_spark(spark, slide_nums, folder="data", training=False, tile_size=1024, overlap=0,
               tissue_threshold=0.9, sample_size=256, grayscale=False, normalize_stains=True,
               num_partitions=20000):

    slides = (spark.sparkContext
              .parallelize(slide_nums)
              .filter(lambda slide: open_slide(slide, folder, training) is not None))
    tile_indices = (slides.flatMap(
        lambda slide: process_slide(slide, folder, training, tile_size, overlap)))
    tile_indices = tile_indices.repartition(num_partitions)
    tile_indices.cache()

    tiles = tile_indices.map(lambda tile_index: process_tile_index(tile_index, folder, training))
    filtered_tiles = tiles.filter(lambda tile: keep_tile(tile, tile_size, tissue_threshold))
    samples = filtered_tiles.flatMap(lambda tile: process_tile(tile, sample_size, grayscale))
    if normalize_stains:
        samples = samples.map(lambda sample: normalize_staining(sample))

    return samples

РЕДАКТИРОВАТЬ: Я использую

PYSPARK_PYTHON=python3 spark-submit --master spark://127.0.1.1:7077 spark_preprocessing.py

для запуска приложения.Похоже, что после действия foreach больше ничего не происходит.Есть ли причина для этого?

Answer 1

Вы можете устранить возникшую проблему, если collect все свои семплы подключите к узлу драйвера, прежде чем пытаться их сохранить.Если вы переопределите save_rdd_2_jpeg как:

def save_rdd_2_jpeg(rdd, save_dir):
    for sample in rdd.collect():
        save_nonlabelled_sample_2_jpeg(sample, save_dir)

, тогда все должно работать.