You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by GitBox <gi...@apache.org> on 2022/10/01 05:58:56 UTC
[GitHub] [spark] chaoqin-li1123 opened a new pull request, #38066: [SPARK-40509][SS][PYTHON] Add example for applyInPandasWithState followup
chaoqin-li1123 opened a new pull request, #38066:
URL: https://github.com/apache/spark/pull/38066
### What changes were proposed in this pull request?
This is a followup of #38013 which introduce an example for applyInPandasWithState. Address some comments on code style.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon closed pull request #38066: [SPARK-40509][SS][PYTHON][FOLLLOW-UP] Add example for applyInPandasWithState followup
Posted by GitBox <gi...@apache.org>.
HyukjinKwon closed pull request #38066: [SPARK-40509][SS][PYTHON][FOLLLOW-UP] Add example for applyInPandasWithState followup
URL: https://github.com/apache/spark/pull/38066
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] chaoqin-li1123 commented on a diff in pull request #38066: [SPARK-40509][SS][PYTHON] Add example for applyInPandasWithState followup
Posted by GitBox <gi...@apache.org>.
chaoqin-li1123 commented on code in PR #38066:
URL: https://github.com/apache/spark/pull/38066#discussion_r985390460
##########
examples/src/main/python/sql/streaming/structured_network_wordcount_session_window.py:
##########
@@ -71,55 +72,57 @@
# Split the lines into words, retaining timestamps, each word become a sessionId
events = lines.select(
explode(split(lines.value, " ")).alias("sessionId"),
- lines.timestamp.cast("long"),
+ lines.timestamp,
)
# Type of output records.
session_schema = StructType(
[
StructField("sessionId", StringType()),
StructField("count", LongType()),
- StructField("start", LongType()),
- StructField("end", LongType()),
+ StructField("start", TimestampType()),
+ StructField("end", TimestampType()),
]
)
# Type of group state.
# Omit the session id in the state since it is available as group key
session_state_schema = StructType(
[
StructField("count", LongType()),
- StructField("start", LongType()),
- StructField("end", LongType()),
+ StructField("start", TimestampType()),
+ StructField("end", TimestampType()),
]
)
def func(
- key: Any, pdf_iter: Iterable[pd.DataFrame], state: GroupState
+ key: Any, pdfs: Iterable[pd.DataFrame], state: GroupState
) -> Iterable[pd.DataFrame]:
if state.hasTimedOut:
count, start, end = state.get
state.remove()
+ (session_id,) = key
yield pd.DataFrame(
{
- "sessionId": [key[0]],
+ "sessionId": [session_id],
"count": [count],
"start": [start],
"end": [end],
}
)
else:
- start = math.inf
- end = 0
- count = 0
+ pdf_iter = iter(pdfs)
Review Comment:
Now start and end are TimestampType, I don't know what should be reasonable min and max values for then
##########
examples/src/main/python/sql/streaming/structured_network_wordcount_session_window.py:
##########
@@ -71,55 +72,57 @@
# Split the lines into words, retaining timestamps, each word become a sessionId
events = lines.select(
explode(split(lines.value, " ")).alias("sessionId"),
- lines.timestamp.cast("long"),
+ lines.timestamp,
)
# Type of output records.
session_schema = StructType(
[
StructField("sessionId", StringType()),
StructField("count", LongType()),
- StructField("start", LongType()),
- StructField("end", LongType()),
+ StructField("start", TimestampType()),
+ StructField("end", TimestampType()),
]
)
# Type of group state.
# Omit the session id in the state since it is available as group key
session_state_schema = StructType(
[
StructField("count", LongType()),
- StructField("start", LongType()),
- StructField("end", LongType()),
+ StructField("start", TimestampType()),
+ StructField("end", TimestampType()),
]
)
def func(
- key: Any, pdf_iter: Iterable[pd.DataFrame], state: GroupState
+ key: Any, pdfs: Iterable[pd.DataFrame], state: GroupState
) -> Iterable[pd.DataFrame]:
if state.hasTimedOut:
count, start, end = state.get
state.remove()
+ (session_id,) = key
yield pd.DataFrame(
{
- "sessionId": [key[0]],
+ "sessionId": [session_id],
"count": [count],
"start": [start],
"end": [end],
}
)
else:
- start = math.inf
- end = 0
- count = 0
+ pdf_iter = iter(pdfs)
Review Comment:
Now start and end are TimestampType, I don't know what should be reasonable initial values for them.
##########
examples/src/main/python/sql/streaming/structured_network_wordcount_session_window.py:
##########
@@ -71,55 +72,57 @@
# Split the lines into words, retaining timestamps, each word become a sessionId
events = lines.select(
explode(split(lines.value, " ")).alias("sessionId"),
- lines.timestamp.cast("long"),
+ lines.timestamp,
)
# Type of output records.
session_schema = StructType(
[
StructField("sessionId", StringType()),
StructField("count", LongType()),
- StructField("start", LongType()),
- StructField("end", LongType()),
+ StructField("start", TimestampType()),
+ StructField("end", TimestampType()),
]
)
# Type of group state.
# Omit the session id in the state since it is available as group key
session_state_schema = StructType(
[
StructField("count", LongType()),
- StructField("start", LongType()),
- StructField("end", LongType()),
+ StructField("start", TimestampType()),
+ StructField("end", TimestampType()),
]
)
def func(
- key: Any, pdf_iter: Iterable[pd.DataFrame], state: GroupState
+ key: Any, pdfs: Iterable[pd.DataFrame], state: GroupState
) -> Iterable[pd.DataFrame]:
if state.hasTimedOut:
count, start, end = state.get
state.remove()
+ (session_id,) = key
yield pd.DataFrame(
{
- "sessionId": [key[0]],
+ "sessionId": [session_id],
"count": [count],
"start": [start],
"end": [end],
}
)
else:
- start = math.inf
- end = 0
- count = 0
+ pdf_iter = iter(pdfs)
Review Comment:
start = min(start, pdf["timestamp"].min())
end = max(end, pdf["timestamp"].max())
These statement expects a non-empty start and end value.
Also, can this callback ever be invoked with empty records dataframe?
##########
examples/src/main/python/sql/streaming/structured_network_wordcount_session_window.py:
##########
@@ -71,55 +72,57 @@
# Split the lines into words, retaining timestamps, each word become a sessionId
events = lines.select(
explode(split(lines.value, " ")).alias("sessionId"),
- lines.timestamp.cast("long"),
+ lines.timestamp,
)
# Type of output records.
session_schema = StructType(
[
StructField("sessionId", StringType()),
StructField("count", LongType()),
- StructField("start", LongType()),
- StructField("end", LongType()),
+ StructField("start", TimestampType()),
+ StructField("end", TimestampType()),
]
)
# Type of group state.
# Omit the session id in the state since it is available as group key
session_state_schema = StructType(
[
StructField("count", LongType()),
- StructField("start", LongType()),
- StructField("end", LongType()),
+ StructField("start", TimestampType()),
+ StructField("end", TimestampType()),
]
)
def func(
- key: Any, pdf_iter: Iterable[pd.DataFrame], state: GroupState
+ key: Any, pdfs: Iterable[pd.DataFrame], state: GroupState
) -> Iterable[pd.DataFrame]:
if state.hasTimedOut:
count, start, end = state.get
state.remove()
+ (session_id,) = key
yield pd.DataFrame(
{
- "sessionId": [key[0]],
+ "sessionId": [session_id],
"count": [count],
"start": [start],
"end": [end],
}
)
else:
- start = math.inf
- end = 0
- count = 0
+ pdf_iter = iter(pdfs)
Review Comment:
It seems that the pandas udf expects Iterable instead of Iterator as input signature, so I can't change the input type annotation. And this conversion is necessary for us to iterate the first element and then traverse the rest. Technically we know pdfs is an iterator and the conversion is unnecessary.(I tested it locally and it worked without conversion). But the linter mypy will complain.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] chaoqin-li1123 commented on pull request #38066: [SPARK-40509][SS][PYTHON] Add example for applyInPandasWithState followup
Posted by GitBox <gi...@apache.org>.
chaoqin-li1123 commented on PR #38066:
URL: https://github.com/apache/spark/pull/38066#issuecomment-1264255378
@HeartSaVioR @HyukjinKwon
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #38066: [SPARK-40509][SS][PYTHON][FOLLLOW-UP] Add example for applyInPandasWithState followup
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on code in PR #38066:
URL: https://github.com/apache/spark/pull/38066#discussion_r985401190
##########
examples/src/main/python/sql/streaming/structured_network_wordcount_session_window.py:
##########
@@ -71,55 +72,57 @@
# Split the lines into words, retaining timestamps, each word become a sessionId
events = lines.select(
explode(split(lines.value, " ")).alias("sessionId"),
- lines.timestamp.cast("long"),
+ lines.timestamp,
)
# Type of output records.
session_schema = StructType(
[
StructField("sessionId", StringType()),
StructField("count", LongType()),
- StructField("start", LongType()),
- StructField("end", LongType()),
+ StructField("start", TimestampType()),
+ StructField("end", TimestampType()),
]
)
# Type of group state.
# Omit the session id in the state since it is available as group key
session_state_schema = StructType(
[
StructField("count", LongType()),
- StructField("start", LongType()),
- StructField("end", LongType()),
+ StructField("start", TimestampType()),
+ StructField("end", TimestampType()),
]
)
def func(
- key: Any, pdf_iter: Iterable[pd.DataFrame], state: GroupState
+ key: Any, pdfs: Iterable[pd.DataFrame], state: GroupState
) -> Iterable[pd.DataFrame]:
if state.hasTimedOut:
count, start, end = state.get
state.remove()
+ (session_id,) = key
yield pd.DataFrame(
{
- "sessionId": [key[0]],
+ "sessionId": [session_id],
"count": [count],
"start": [start],
"end": [end],
}
)
else:
- start = math.inf
- end = 0
- count = 0
+ pdf_iter = iter(pdfs)
Review Comment:
Hm .. this is actually a general issue to fix. Okay, LGTM
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on a diff in pull request #38066: [SPARK-40509][SS][PYTHON] Add example for applyInPandasWithState followup
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on code in PR #38066:
URL: https://github.com/apache/spark/pull/38066#discussion_r985388652
##########
examples/src/main/python/sql/streaming/structured_network_wordcount_session_window.py:
##########
@@ -71,55 +72,57 @@
# Split the lines into words, retaining timestamps, each word become a sessionId
events = lines.select(
explode(split(lines.value, " ")).alias("sessionId"),
- lines.timestamp.cast("long"),
+ lines.timestamp,
)
# Type of output records.
session_schema = StructType(
[
StructField("sessionId", StringType()),
StructField("count", LongType()),
- StructField("start", LongType()),
- StructField("end", LongType()),
+ StructField("start", TimestampType()),
+ StructField("end", TimestampType()),
]
)
# Type of group state.
# Omit the session id in the state since it is available as group key
session_state_schema = StructType(
[
StructField("count", LongType()),
- StructField("start", LongType()),
- StructField("end", LongType()),
+ StructField("start", TimestampType()),
+ StructField("end", TimestampType()),
]
)
def func(
- key: Any, pdf_iter: Iterable[pd.DataFrame], state: GroupState
+ key: Any, pdfs: Iterable[pd.DataFrame], state: GroupState
) -> Iterable[pd.DataFrame]:
if state.hasTimedOut:
count, start, end = state.get
state.remove()
+ (session_id,) = key
yield pd.DataFrame(
{
- "sessionId": [key[0]],
+ "sessionId": [session_id],
"count": [count],
"start": [start],
"end": [end],
}
)
else:
- start = math.inf
- end = 0
- count = 0
+ pdf_iter = iter(pdfs)
Review Comment:
Hm, why should we do this? I think you can directly iterate `pdfs` over.
Also, I think the previous value initialization logic looked making sense. This example now would fail if the function is executed with an empty input.
##########
examples/src/main/python/sql/streaming/structured_network_wordcount_session_window.py:
##########
@@ -71,55 +72,57 @@
# Split the lines into words, retaining timestamps, each word become a sessionId
events = lines.select(
explode(split(lines.value, " ")).alias("sessionId"),
- lines.timestamp.cast("long"),
+ lines.timestamp,
)
# Type of output records.
session_schema = StructType(
[
StructField("sessionId", StringType()),
StructField("count", LongType()),
- StructField("start", LongType()),
- StructField("end", LongType()),
+ StructField("start", TimestampType()),
+ StructField("end", TimestampType()),
]
)
# Type of group state.
# Omit the session id in the state since it is available as group key
session_state_schema = StructType(
[
StructField("count", LongType()),
- StructField("start", LongType()),
- StructField("end", LongType()),
+ StructField("start", TimestampType()),
+ StructField("end", TimestampType()),
]
)
def func(
- key: Any, pdf_iter: Iterable[pd.DataFrame], state: GroupState
+ key: Any, pdfs: Iterable[pd.DataFrame], state: GroupState
) -> Iterable[pd.DataFrame]:
if state.hasTimedOut:
count, start, end = state.get
state.remove()
+ (session_id,) = key
yield pd.DataFrame(
{
- "sessionId": [key[0]],
+ "sessionId": [session_id],
"count": [count],
"start": [start],
"end": [end],
}
)
else:
- start = math.inf
- end = 0
- count = 0
+ pdf_iter = iter(pdfs)
Review Comment:
I thin you can just saw it as `None`
##########
examples/src/main/python/sql/streaming/structured_network_wordcount_session_window.py:
##########
@@ -71,55 +72,57 @@
# Split the lines into words, retaining timestamps, each word become a sessionId
events = lines.select(
explode(split(lines.value, " ")).alias("sessionId"),
- lines.timestamp.cast("long"),
+ lines.timestamp,
)
# Type of output records.
session_schema = StructType(
[
StructField("sessionId", StringType()),
StructField("count", LongType()),
- StructField("start", LongType()),
- StructField("end", LongType()),
+ StructField("start", TimestampType()),
+ StructField("end", TimestampType()),
]
)
# Type of group state.
# Omit the session id in the state since it is available as group key
session_state_schema = StructType(
[
StructField("count", LongType()),
- StructField("start", LongType()),
- StructField("end", LongType()),
+ StructField("start", TimestampType()),
+ StructField("end", TimestampType()),
]
)
def func(
- key: Any, pdf_iter: Iterable[pd.DataFrame], state: GroupState
+ key: Any, pdfs: Iterable[pd.DataFrame], state: GroupState
) -> Iterable[pd.DataFrame]:
if state.hasTimedOut:
count, start, end = state.get
state.remove()
+ (session_id,) = key
yield pd.DataFrame(
{
- "sessionId": [key[0]],
+ "sessionId": [session_id],
"count": [count],
"start": [start],
"end": [end],
}
)
else:
- start = math.inf
- end = 0
- count = 0
+ pdf_iter = iter(pdfs)
Review Comment:
Hm, when timeout is reached, it goes to the above branch so technically it won't be. Okay, LGTM but can we avoid `iter(pdfs)` conversion? It's already an iterator.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
[GitHub] [spark] HyukjinKwon commented on pull request #38066: [SPARK-40509][SS][PYTHON][FOLLLOW-UP] Add example for applyInPandasWithState followup
Posted by GitBox <gi...@apache.org>.
HyukjinKwon commented on PR #38066:
URL: https://github.com/apache/spark/pull/38066#issuecomment-1264971134
Merged to master.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org