Skip to content
This repository was archived by the owner on Apr 15, 2024. It is now read-only.

FAB-6528 update spark streaming query listener to use QueryIdleEvent #26

Open
mcriscolo-cs wants to merge 3 commits into
base: main
Choose a base branch
from
Open

FAB-6528 update spark streaming query listener to use QueryIdleEvent #26

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
327e6d6
FAB-6528 update spark streaming query listener to use QueryIdleEvent
mcriscolo-cs Mar 4, 2024
0143cd2
FAB-6528 update spark streaming query listener to use QueryIdleEvent
mcriscolo-cs Mar 4, 2024
b73e043
FAB-6498 Delta Lake 3.1.0 upgrade
mcriscolo-cs Feb 12, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
31 changes: 17 additions & 14 deletions catalog-management/src/main/java/com/c12e/cortex/examples/catalog/RailedCommand.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -257,24 +257,27 @@ public void onQueryStarted(QueryStartedEvent event) {
}

@Override
public void onQueryProgress(QueryProgressEvent event) {
logger.info("STREAMING LISTENER: Streaming Query in progress");
if (event.progress().numInputRows() == 0) {
countBeforeStop--;
if (countBeforeStop == 0) {
logger.info("STREAMING LISTENER: Initiating Streaming Query stop");
try {
sparkSession.sqlContext().streams().get(event.progress().id()).stop();
countBeforeStop = 1L;
} catch (TimeoutException e) {
logger.error("STREAMING LISTENER: Timeout error in query", e);
}
public void onQueryIdle(QueryIdleEvent event) {
logger.info("STREAMING LISTENER: Streaming Query idle");
countBeforeStop--;
if(countBeforeStop == 0){
logger.info("STREAMING LISTENER: Initiating Streaming Query stop");
try {
sparkSession.sqlContext().streams().get(event.id()).stop();
} catch (TimeoutException e) {
logger.error("Timeout error in query", e);
}
} else {
logger.info("STREAMING LISTENER: No processing occurred in last poll, stopping in {} poll intervals", countBeforeStop);
}
logger.info(event.progress().prettyJson());
logger.info("STREAMING LISTENER: No processing occurred in last poll, stopping in {} poll intervals", countBeforeStop);
}

@Override
public void onQueryProgress(QueryProgressEvent event) {
logger.info("STREAMING LISTENER: Streaming Query in progress");
logger.info(event.progress().prettyJson());
}

@Override
public void onQueryTerminated(QueryTerminatedEvent event) {
logger.info("STREAMING LISTENER: onQueryTerminated");
Expand Down
28 changes: 16 additions & 12 deletions ...ource-streaming/src/main/java/com/c12e/cortex/examples/streaming/StreamingDataSource.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -105,21 +105,25 @@ public void onQueryStarted(QueryStartedEvent event) {
}

@Override
public void onQueryProgress(QueryProgressEvent event) {
logger.info("STREAMING LISTENER: Streaming Query in progress");
if (event.progress().numInputRows() == 0) {
countBeforeStop--;
if(countBeforeStop == 0){
logger.info("STREAMING LISTENER: Initiating Streaming Query stop");
try {
sparkSession.sqlContext().streams().get(event.progress().id()).stop();
} catch (TimeoutException e) {
logger.error("STREAMING LISTENER: Timeout error in query", e);
}
public void onQueryIdle(QueryIdleEvent event) {
logger.info("STREAMING LISTENER: Streaming Query idle");
countBeforeStop--;
if(countBeforeStop == 0){
logger.info("STREAMING LISTENER: Initiating Streaming Query stop");
try {
sparkSession.sqlContext().streams().get(event.id()).stop();
} catch (TimeoutException e) {
logger.error("Timeout error in query", e);
}
} else {
logger.info("STREAMING LISTENER: No processing occurred in last poll, stopping in {} poll intervals", countBeforeStop);
}
}

@Override
public void onQueryProgress(QueryProgressEvent event) {
logger.info("STREAMING LISTENER: Streaming Query in progress");
logger.info(event.progress().prettyJson());
logger.info("STREAMING LISTENER: No processing occurred in last poll, stopping in {} poll intervals", countBeforeStop);
}

@Override
Expand Down
2 changes: 1 addition & 1 deletion main-app/src/main/resources/python/submit_job.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,7 +33,7 @@ def get_runtime_args(config):
args.append(key)
args.append("{}={}".format(y, s_val))
args.append('--py-files')
args.append(f'local:///opt/spark/jars/profiles-sdk-{os.environ["VERSION"]}.jar,local:///opt/spark/jars/delta-spark_2.12-3.0.0.jar')
args.append(f'local:///opt/spark/jars/profiles-sdk-{os.environ["VERSION"]}.jar,local:///opt/spark/jars/delta-spark_2.12-3.1.0.jar')
args.append(pyspark_args['app_location'])
for x in pyspark_args['app_command']:
args.append(x)
Expand Down