#1616 Add default spark image Pull request opened by pingsutw TL;DR New image size 3.2 GB vs Old image size 5.13 GB

24K    ./lib/python3.9/site-packages/pyspark/sql/avro
 12K    ./lib/python3.9/site-packages/pyspark/sql/pandas/_typing/protocols
 24K    ./lib/python3.9/site-packages/pyspark/sql/pandas/_typing
104K    ./lib/python3.9/site-packages/pyspark/sql/pandas/__pycache__
264K    ./lib/python3.9/site-packages/pyspark/sql/pandas
1.6M    ./lib/python3.9/site-packages/pyspark/sql
301M    ./lib/python3.9/site-packages/pyspark

=====================================

• Add a docker file in

/flytekitplugins-spark

• Update GH workflow to build spark image • Use default executor path and applications path when using default spark image Type ☐ Bug Fix ☑︎ Feature ☐ Plugin Are all requirements met? ☑︎ Code completed ☑︎ Smoke tested ☐ Unit tests added ☐ Code documentation added ☐ Any pending items have an associated Issue Complete description

import datetime
import random
from operator import add

import flytekit
from flytekit import Resources, task, workflow
from flytekitplugins.spark import Spark
from flytekit.image_spec.image_spec import ImageSpec

spark_image = ImageSpec(registry="pingsutw")


@task(
    task_config=Spark(
        # this configuration is applied to the spark cluster
        spark_conf={
            "spark.driver.memory": "1000M",
            "spark.executor.memory": "1000M",
            "spark.executor.cores": "1",
            "spark.executor.instances": "2",
            "spark.driver.cores": "1",
        }
    ),
    limits=Resources(mem="2000M"),
    cache_version="1",
    container_image=spark_image,
)
def hello_spark(partitions: int) -> float:
    print("Starting Sparkfk wifth Partitions: {}".format(partitions))
    n = 100000 * partitions
    sess = flytekit.current_context().spark_session
    count = (
        sess.sparkContext.parallelize(range(1, n + 1), partitions).map(f).reduce(add)
    )
    pi_val = 4.0 * count / n
    print("Pi val is :{}".format(pi_val))
    return pi_val


def f(_):
    x = random.random() * 2 - 1
    y = random.random() * 2 - 1
    return 1 if x**2 + y**2 <= 1 else 0


@task(cache_version="1", container_image=spark_image)
def print_every_time(value_to_print: float, date_triggered: datetime.datetime) -> int:
    print("My printed value: {} @ {}".format(value_to_print, date_triggered))
    return 1


@workflow
def wf(triggered_date: datetime.datetime = datetime.datetime.now()) -> float:
    """
    Using the workflow is still as any other workflow. As image is a property of the task, the workflow does not care
    about how the image is configured.
    """
    pi = hello_spark(partitions=50)
    print_every_time(value_to_print=pi, date_triggered=triggered_date)
    return pi


if __name__ == "__main__":
    print(f"Running {__file__} main...")
    print(
        f"Running my_spark(triggered_date=datetime.datetime.now()){wf(triggered_date=datetime.datetime.now())}"
    )

Tracking Issue NA Follow-up issue NA flyteorg/flytekit Codecov: 54.54% of diff hit (target 70.10%) Codecov: 70.10% (-0.01%) compared to baf0b89 ✅ 28 other checks have passed 28/30 successful checks