火花:如何在VS代码中调试熊猫-UDF

Question

我正在寻找一种在vscode和Pycharm社区版本(place断点并停止在UDF中)中调试火花熊猫UDF的方法。当断点放置在UDF调试器中时，UDF调试器不会停止。

在下面的参考文献中，描述了本地模式和分布式模式。

我正在尝试至少在本地模式下进行调试。Pycharm/VS代码应该有一种通过“附加到本地进程”来调试本地enc的方法。只是我想不出是怎么回事。

目前，我还没有找到任何答案，如何在VS代码(我的dev ide)中将pyspark调试器附加到UDF内部的本地进程。

我在Pycharm中只找到了下面的例子。

本地进程How can PySpark be called in debug mode?的

1. 附加程序

当我试图附加到进程时，我在Pycharm中得到了下面的消息。在VS代码中，我得到了进程不能附加的消息。

Attaching to a process with PID=33,692
/home/usr_name/anaconda3/envs/yf/bin/python3.8 /snap/pycharm-community/223/plugins/python-ce/helpers/pydev/pydevd_attach_to_process/attach_pydevd.py --port 40717 --pid 33692
WARNING: The 'kernel.yama.ptrace_scope' parameter value is not 0, attach to process may not work correctly.
         Please run 'sudo sysctl kernel.yama.ptrace_scope=0' to change the value temporary
     or add the 'kernel.yama.ptrace_scope = 0' line to /etc/sysctl.d/10-ptrace.conf to set it permanently.

Process finished with exit code 0
Server stopped.

1. pyspark_xray https://github.com/bradyjiang/pyspark_xray与此包一起使用，可以调试在worker上运行的rdd，但我无法调整包来调试UDFs

。

示例代码中，断点并不停留在UDF pandas_function(Url_json)中：

import pandas as pd
import pyspark
from pyspark.sql import Row
from pyspark.sql.types import StructType, StructField, IntegerType,StringType
spark = pyspark.sql.SparkSession.builder.appName("test") \
    .master('local[*]') \
    .getOrCreate()
sc = spark.sparkContext

# Create initial dataframe respond_sdf
d_list = [('api_1',"{'api': ['api_1', 'api_1', 'api_1'],'A': [1,2,3], 'B': [4,5,6] }"),
            (' api_2', "{'api': ['api_2', 'api_2', 'api_2'],'A': [7,8,9], 'B': [10,11,12] }")]

schema = StructType([
  StructField('url', StringType(), True),
  StructField('content', StringType(), True)
  ])

jsons = sc.parallelize(rdd_list)
respond_sdf = spark.createDataFrame(jsons, schema)

# Pandas UDF 
def pandas_function(url_json):
# Here I want to place breakpoint
    df = pd.DataFrame(eval(url_json['content'][0]))
    return df

# Pnadas UDF transformation applied to respond_sdf
respond_sdf.groupby(F.monotonically_increasing_id()).applyInPandas(pandas_function, schema=schema).show()

Answer 1

此示例演示如何使用优秀的pyspark_exray库逐步进入传递到Dataframe.mapInPandas函数的UDF函数。

https://github.com/bradyjiang/pyspark_xray/blob/master/demo_app02/driver.py