Anyone able to replicate the pyspark example demo?

[zhh210]

I was following the example of running a customized spark.
It was fine while executing python test.py. However, when using a manual run to materialize an asset (from dagster dev -f test.py ), I got the error of:

The above exception was caused by the following exception:
pyspark.errors.exceptions.base.PySparkRuntimeError: [CONTEXT_ONLY_VALID_ON_DRIVER] It appears that you are attempting to reference SparkContext from a broadcast variable, action, or transformation. SparkContext can only be used on the driver, not in code that it run on workers. For more information, see SPARK-5063.
File "/opt/conda/lib/python3.9/site-packages/dagster/_core/execution/plan/utils.py", line 54, in op_execution_error_boundary
yield
File "/opt/conda/lib/python3.9/site-packages/dagster/_utils/init.py", line 465, in iterate_with_context
next_output = next(iterator)
File "/opt/conda/lib/python3.9/site-packages/dagster/_core/execution/plan/execute_step.py", line 724, in _gen_fn
gen_output = output_manager.handle_output(output_context, output.value)
File "/opt/conda/lib/python3.9/site-packages/dagster/_core/storage/upath_io_manager.py", line 453, in handle_output
self.dump_to_path(context=context, obj=obj, path=path)
File "/opt/conda/lib/python3.9/site-packages/dagster/_core/storage/fs_io_manager.py", line 260, in dump_to_path
pickle.dump(obj, file, PICKLE_PROTOCOL)
File "/opt/conda/lib/python3.9/site-packages/pyspark/context.py", line 466, in getnewargs
raise PySparkRuntimeError(

Any idea on how I can address this error? Thanks.

[salazarm]

cc @tacastillo

[zhh210]

Adding more details to the issue here. What I'm trying to achieve is using a customized spark cluster as resource, and creating two assets where each is a pyspark DataFrame but only one will be saved as parquet.

Here's my code:

from pathlib import Path
from typing import Any

from dagster import ConfigurableIOManager, Definitions, ResourceParam, asset, graph
# from dagster_aws.emr import emr_pyspark_step_launcher
# from dagster_aws.s3 import S3Resource
# from dagster_pyspark import PySparkResource
from pyspark.sql import DataFrame, Row
from pyspark.sql.types import IntegerType, StringType, StructField, StructType

from dagster._core.definitions.no_step_launcher import no_step_launcher

### The following lines are from customized pyspark creation
import os,sys
src_folder = os.path.abspath("../")
print(src_folder)
sys.path.append(src_folder)
from clustering.utils.deepsea_spark import create_deepsea_spark, get_spark_conf
from clustering.utils.deepsea_resource import pyspark_resource, PySparkResource
spark_conf = get_spark_conf(spark_yaml = "configs/conf_spark.yaml")
emr_pyspark = PySparkResource(spark_config = spark_conf)
print(type(emr_pyspark))

class ParquetIOManager(ConfigurableIOManager):
    pyspark: PySparkResource
    path_prefix: str

    def _get_path(self, context) -> str:
        return "/".join([context.resource_config["path_prefix"], *context.step_key]) # have to replace asset_key by step_key otherwise dagster will complain asset_key doesn't exist

    def handle_output(self, context, obj):
        obj.write.parquet(self._get_path(context))

    def load_input(self, context):
        spark = self.pyspark.spark_session
        return spark.read.parquet(self._get_path(context.upstream_output))


@asset(required_resource_keys={'pyspark', 'pyspark_step_launcher'})
def people(context) -> DataFrame:
    schema = StructType([StructField("name", StringType()), StructField("age", IntegerType())])
    rows = [Row(name="Thom", age=51), Row(name="Jonny", age=48), Row(name="Nigel", age=49)]
    return context.resources.pyspark.spark_session.createDataFrame(rows, schema)


# emr_pyspark = PySparkResource(spark_config={"spark.executor.memory": "2g"})


@asset(deps=[people], required_resource_keys={"pyspark_step_launcher"},io_manager_key="io_manager_materialize_final")
def people_over_50(context) -> DataFrame:
    return people.filter(people["age"] > 50)

from dagster import mem_io_manager, in_process_executor
local_resource_defs = {
        "pyspark_step_launcher": no_step_launcher,
        "pyspark": emr_pyspark,
        # "s3": S3Resource(),
        "io_manager_materialize_final": ParquetIOManager(pyspark=emr_pyspark, path_prefix="s3://my-testing-bucket"),
    }

defs = Definitions(
    assets=[people, people_over_50],
    resources=local_resource_defs,
)

res = defs.get_implicit_global_asset_job_def()#.execute_in_process()



@graph
def make_and_filter_data():
    return people_over_50(people())

pyspark_sample_job_local = make_and_filter_data.to_job(
        name="pyspark_sample_job_local", resource_defs=local_resource_defs,
        executor_def=in_process_executor
    )

pyspark_sample_job_local.execute_in_process()

With the failure log:

dagster._core.errors.DagsterExecutionHandleOutputError: Error occurred while handling output "result" of step "people":

  File "/opt/conda/lib/python3.9/site-packages/dagster/_core/execution/plan/execute_plan.py", line 275, in dagster_event_sequence_for_step
    for step_event in check.generator(step_events):
  File "/opt/conda/lib/python3.9/site-packages/dagster/_core/execution/plan/execute_step.py", line 485, in core_dagster_event_sequence_for_step
    for evt in _type_check_and_store_output(step_context, user_event):
  File "/opt/conda/lib/python3.9/site-packages/dagster/_core/execution/plan/execute_step.py", line 536, in _type_check_and_store_output
    for evt in _store_output(step_context, step_output_handle, output):
  File "/opt/conda/lib/python3.9/site-packages/dagster/_core/execution/plan/execute_step.py", line 734, in _store_output
    for elt in iterate_with_context(
  File "/opt/conda/lib/python3.9/site-packages/dagster/_utils/__init__.py", line 467, in iterate_with_context
    return
  File "/opt/conda/lib/python3.9/contextlib.py", line 137, in __exit__
    self.gen.throw(typ, value, traceback)
  File "/opt/conda/lib/python3.9/site-packages/dagster/_core/execution/plan/utils.py", line 84, in op_execution_error_boundary
    raise error_cls(

The above exception was caused by the following exception:
pyspark.errors.exceptions.base.PySparkRuntimeError: [CONTEXT_ONLY_VALID_ON_DRIVER] It appears that you are attempting to reference SparkContext from a broadcast variable, action, or transformation. SparkContext can only be used on the driver, not in code that it run on workers. For more information, see SPARK-5063.

  File "/opt/conda/lib/python3.9/site-packages/dagster/_core/execution/plan/utils.py", line 54, in op_execution_error_boundary
    yield
  File "/opt/conda/lib/python3.9/site-packages/dagster/_utils/__init__.py", line 465, in iterate_with_context
    next_output = next(iterator)
  File "/opt/conda/lib/python3.9/site-packages/dagster/_core/execution/plan/execute_step.py", line 724, in _gen_fn
    gen_output = output_manager.handle_output(output_context, output.value)
  File "/opt/conda/lib/python3.9/site-packages/dagster/_core/storage/upath_io_manager.py", line 453, in handle_output
    self.dump_to_path(context=context, obj=obj, path=path)
  File "/opt/conda/lib/python3.9/site-packages/dagster/_core/storage/fs_io_manager.py", line 260, in dump_to_path
    pickle.dump(obj, file, PICKLE_PROTOCOL)
  File "/opt/conda/lib/python3.9/site-packages/pyspark/context.py", line 466, in __getnewargs__
    raise PySparkRuntimeError(

[zhh210]

Another interesting behavior: The op and graph example of this link works when I use python test.py but shows an empty graph in dagster dev -f test.py.