<jupyter_start><jupyter_text>Land image data into Delta Table with Spark<jupyter_code>import pyspark
spark = pyspark.sql.SparkSession.builder.appName("MyApp") \
    .config("spark.jars.packages", "io.delta:delta-core_2.11:0.6.0") \
    .getOrCreate()
sc = spark.sparkContext
sc.addPyFile("/usr/lib/spark/jars/delta-core_2.11-0.6.0.jar")
from delta.tables import *

RAW_DATA_BUCKET = ""

import boto3
import os

def load_voc_image_names(bucket_name, prefix):
    s3 = boto3.resource('s3')
    bucket = s3.Bucket(bucket_name)
    files = []
    for obj in bucket.objects.filter(Prefix=os.path.join(prefix, "JPEGImages")):
        if obj.key.endswith('.jpg'):
            files.append(obj.key)
    return(files)
    
V1_DATA_FOLDER = "v1"
keys = load_voc_image_names(RAW_DATA_BUCKET, V1_DATA_FOLDER)

import io
import random
random.seed(42)

def load_image_list(bucket_name, prefix, list_name):
    s3 = boto3.resource('s3')
    bucket = s3.Bucket(bucket_name)
    path = os.path.join(prefix, "ImageSets", "Main", f"{list_name}.txt")
    response = bucket.Object(path)
    data = response.get()['Body'].read()
    return data.decode('utf8').split('\n')


train_list = load_image_list(RAW_DATA_BUCKET, V1_DATA_FOLDER, 'train')
val_list = load_image_list(RAW_DATA_BUCKET, V1_DATA_FOLDER, 'val')
random.shuffle(val_list)
val_list = val_list[:1000]
train_files = [k for k in keys if os.path.basename(k).split('.')[0] in train_list]
val_files = [k for k in keys if os.path.basename(k).split('.')[0] in val_list]
print(f"{len(train_files)} files in training dataset")
print(f"{len(val_files)} files in val dataset")
def download_from_s3(row):
    s3 = boto3.client('s3')
    bucket = RAW_DATA_BUCKET
    key = row.image_key  
    filename = os.path.basename(str(key))
    basename = filename.split('.')[0]
    response = s3.get_object(Bucket=bucket, Key=key)
    body = response["Body"]
    contents = bytearray(body.read())
    body.close()
    
    annotation_key = key.replace("JPEGImages", "Annotations").replace('.jpg', '.xml')
    response = s3.get_object(Bucket=bucket, Key=annotation_key)
    data = response['Body'].read()
    annotations = data.decode('utf8')

    if len(contents):
        return (key, contents, annotations)
from pyspark.sql.types import StructType, StructField, IntegerType, BinaryType, StringType
from pyspark.sql import Row

def write_to_delta(files, bucket, table, append=False):
    rdd1 = sc.parallelize(files)
    row_rdd = rdd1.map(lambda x: Row(x))
    rows_df = sqlContext.createDataFrame(row_rdd,['image_key'])
    images_rdd = (
      rows_df
      .rdd
      .map(download_from_s3)
    )

    schema = StructType([StructField("key", StringType(), False),
                         StructField("image", BinaryType(), False),
                         StructField("annotations", StringType(), False)]
                        )

    image_df = (
      images_rdd
      .toDF(schema)
    )
    
    if append:
        mode = "append"
    else:
        mode = "overwrite"
    
    (
      image_df
      .write
      .format("delta")
      .mode(mode)
      .option("compression", "gzip")
      .save(f"s3://{bucket}/{table}")
    )

DELTA_BUCKET = ""
TRAIN_TABLE_NAME = "train"
VAL_TABLE_NAME = "val"
write_to_delta(train_files, DELTA_BUCKET, TRAIN_TABLE_NAME)
write_to_delta(val_files, DELTA_BUCKET, VAL_TABLE_NAME)
train_df = spark.read.format("delta").option("versionAsOf", 0).load(f"s3://{DELTA_BUCKET}/{TRAIN_TABLE_NAME}/")
train_df.show(10)<jupyter_output><empty_output><jupyter_text>Update our Delta Table with new images<jupyter_code>updated_keys = load_voc_image_names(RAW_DATA_BUCKET, 'v2')
train_list2 = load_image_list(RAW_DATA_BUCKET, 'v2', 'train')
train_files2 = [k for k in updated_keys if os.path.basename(k).split('.')[0] in train_list2]
print(f"{len(train_files2)} files in the new training dataset")
write_to_delta(train_files2, DELTA_BUCKET, 'train', append=True)<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.1415884293326567, 'fraction_numerical': 0.014970819588936818, 'mean_word_length': 3.977272727272727, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 9, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 1}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 2, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '2', 'dir': 'jupyter-structured-clean-dedup', 'id': '625146', 'license': 'Apache-2.0', 'n_tokens_mistral': 1459, 'n_tokens_neox': 1431, 'n_words': 253, 'path': 'spark_ecosystem/spark/voc_to_delta.ipynb', 'repo_name': 'hoanghphan/works-with-determined', 'size': '3941'}