I have been working with PySpark and distributed computing to do work with dataframes that involve querying PI I have been working with the User Defined functions and have managed to get a semi working code. However, I am getting this socket timeout error, and cannot seem to trace back what the root cause of it is, because it triggers at different points in the program at different times of execution. On top of all of that the program has the occasional successful run, in which no errors are thrown...
Running this using pytest and calling -s -k test_main.py
Here is the error:
for tag in tags:
# print(tag)
total=0
parts = []
parts.append(tag)
result_df = summaries(
spark,
name,
parts,
start_time,
end_time,
SummaryType.COUNT,
CalculationBasis.EVENT_WEIGHTED,
TimestampCalculation.AUTO,
ZoneInfo("America/New_York"),
None # Pass your DataType if needed
)
print(result_df)
> back = result_df.collect()
test_main.py:142:
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
..\..\..\AppData\Local\Programs\Python\Python39-32\lib\site-packages\pyspark\sql\dataframe.py:1261: in collect
sock_info = self._jdf.collectToPython()
..\..\..\AppData\Local\Programs\Python\Python39-32\lib\site-packages\py4j\java_gateway.py:1322: in __call__
return_value = get_return_value(
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _
a = ('xro469', <py4j.clientserver.JavaClient object at 0x044D4730>, 'o465', 'collectToPython'), kw = {}
converted = PythonException('\n An exception was thrown from the Python worker. Please see the stack trace below.\nTraceback (mos...r\n\tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635)\r\n\t... 1 more\r\n')
def deco(*a: Any, **kw: Any) -> Any:
try:
return f(*a, **kw)
except Py4JJavaError as e:
converted = convert_exception(e.java_exception)
if not isinstance(converted, UnknownException):
# Hide where the exception came from that shows a non-Pythonic
# JVM exception message.
> raise converted from None
E pyspark.errors.exceptions.captured.PythonException:
E An exception was thrown from the Python worker. Please see the stack trace below.
E Traceback (most recent call last):
E File "C:\Users\CRAGUM\AppData\Local\Programs\Python\Python39-32\lib\socket.py", line 707, in readinto
E raise
E socket.timeout: timed out
..\..\..\AppData\Local\Programs\Python\Python39-32\lib\site-packages\pyspark\errors\exceptions\captured.py:185: PythonException
I have looked at a lot of other posts to see if any of the issues there could help resolve, but to no avail
Im using Spark version 3.5.1, Python 3.9.0 and Java JDK 17
I have tried all these variations when configuring my spark:
spark = SparkSession.builder \
.appName("ExampleSparkInput") \
.config("spark.network.timeout", "10000000s") \
.config("spark.executor.heartbeatInterval", "10000s") \
.config("spark.task.maxFailures", "100") \
.config("spark.executor.memory", "4g") \
.config("spark.driver.memory", "4g") \
.config("spark.default.parellelism", 4) \
.config("spark.sql.shuffle.partitions", 4) \
.getOrCreate()
As well as testing each of my User Defined Functions separately
def get_average(session: SparkSession, tags: list[str], pi_server: str, start_time: datetime, end_time: datetime, datatype: DataType, interval: timedelta):
"""
Returns a List of batches, tag dictionary with WebIDs, and the datatype to be used in the schema (if applicable)
"""
parts = []
for i in range(0, len(tags), 10):
parts.append(tags[i:i + 10])
tag_df = session.createDataFrame([], schema=initial_schema)
for part in parts:
row = [part,]
tag_df = tag_df.union(session.createDataFrame([row], schema=initial_schema))
tag_df = tag_df.withColumn("WebId", web_id(lit(pi_server), tag_df["Tag"])).drop("Tag")
tag_df = tag_df.withColumn("explodedWebID", explode(from_json("WebId", web_id_schema)))
type_df = tag_df.drop("WebId").withColumn("Type", col("explodedWebID.Type")).drop("explodedWebID")
data_type = DataType()
if datatype is not None:
data_type = datatype
else:
data_type = get_tag_type(type_df)
tag_df = tag_df.withColumn("Freq", tag_freq(tag_df["WebId"])).drop("WebId")
tag_df = tag_df.withColumn("Freqs", explode(from_json("Freq", frequency_schema))).drop("Freq")
tag_df = tag_df.withColumn("WebID", col("Freqs.WebID")) \
.withColumn("Freq", col("Freqs.Frequency")) \
.withColumn("Tag", col("Freqs.Tag")) \
.drop("Freqs")
freqs = tag_df.collect() # It can throw the socket timeout error here
tag_dict = {}
for row in freqs:
web_id_done = row["WebID"]
freq = row["Freq"]
tag = row["Tag"]
tag_dict[tag] = {"Frequency": freq, "WebId": web_id_done}
batches = get_batches(tags, start_time, end_time, interval, tag_dict)
return batches, tag_dict, data_type
As note here is the tagfreq UDF
def get_tagfreq_sync(points: str):
points = json.loads(points)
url = "piwebapi host url"
headers = {
# auth
}
batch_request = {}
end_time = datetime.datetime.now()
start_time_iso = (end_time - datetime.timedelta(days=365)).isoformat() + "Z"
end_time_iso = end_time.isoformat() + "Z"
params = {"startTime": start_time_iso, "endTime": end_time_iso, "summaryType": "Count", "calculationBasis": "EventWeighted", "sampleType": "ExpressionRecordedValues"}
params = '&'.join([f"{key}={value}" for key, value in params.items()])
for item in points:
web_ID = item["WebID"]
batch_request_string = {
"Method": "GET",
"Resource": f"piwebapi host url{web_ID}/summary?{params}"
}
batch_request[item["WebID"]] = batch_request_string
# print(batch_request_string)
with requests.Session() as session:
response = session.post(url, headers=headers, json=batch_request, verify=False)
data = response.json()
flattened_data = []
for item in data:
item_data = data[item]
web_id_for_tag = [point["Tag"] for point in points if point["WebID"] == str(item)][0].lower()
if item_data["Content"]["Items"][0]["Value"]["Value"] == 0:
print("No data for " + item + ". Skipping and setting to 60s")
flattened_data.append({"WebID": item, "Frequency": timedelta_to_iso(datetime.timedelta(seconds=60)), "Tag": web_id_for_tag})
continue
flattened_data.append({"WebID": str(item), \
"Frequency": timedelta_to_iso(datetime.timedelta(days=365) / item_data["Content"]["Items"][0]["Value"]["Value"]), \
"Tag": web_id_for_tag})
# print("FREQTag: " + web_id_for_tag + " Frequency: " + str(datetime.timedelta(days=365) / item_data["Content"]["Items"][0]["Value"]["Value"]))
# print(flattened_data)
return json.dumps(flattened_data)
and as a matter of fact, I have also had it do the socket timeout when calling the get_type function
def get_tag_type(frame: DataFrame):
"""
If the datatype isnt set, this gets the type from all of the rows, and checks to make sure they are all one type of tag.
"""
distinct_values = frame.select("Type").distinct().collect()
if len(distinct_values) != 1:
raise ValueError("Column 'Type' should contain a single unique value in all rows.")
unique_value = distinct_values[0][0]
if "string" in unique_value.lower():
return StringType()
elif "int" in unique_value.lower():
return IntegerType()
elif "float" in unique_value.lower():
return FloatType()
elif "digital" in unique_value.lower():
return StringType()
elif "timestamp" in unique_value.lower():
return StringType()
else:
raise ValueError("Invalid value found in 'types' column.")
Some of the fact that it times out on this is surprising, and its not initially obvious either. I also do not know how much code to include, because some of the functions can be pretty lengthy.
