Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add spark applicationId #1778

Merged
merged 13 commits into from
Jul 5, 2023
Merged

Add spark applicationId #1778

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
9c988b0
Initial spark application draft
rupal-bq Jun 8, 2023
e234a51
Remove temp table
rupal-bq Jun 8, 2023
25160fd
Add license header
rupal-bq Jun 8, 2023
6f95cf0
Add scalastyle-config and update readme
rupal-bq Jun 8, 2023
7e02160
Fix datatype for result and schema
rupal-bq Jun 9, 2023
b72c5a3
Add test
rupal-bq Jun 9, 2023
1883b25
Simplify code using toJSON.collect.toList
rupal-bq Jun 12, 2023
3261028
Add example in readme
rupal-bq Jun 12, 2023
f815709
Fix triple quotes issue
rupal-bq Jun 12, 2023
ec08d0b
Update method name and description
rupal-bq Jun 14, 2023
65e0b1c
Merge branch 'main' into spark-application
rupal-bq Jun 23, 2023
959d5b2
Add applicationId
rupal-bq Jun 25, 2023
82c70f2
Merge branch 'main' into spark-application
rupal-bq Jun 25, 2023
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
4 changes: 3 additions & 1 deletion spark-sql-application/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@
This application execute sql query and store the result in OpenSearch index in following format
```
"stepId":"<emr-step-id>",
"applicationId":"<spark-application-id>"
"schema": "json blob",
"result": "json blob"
```
Expand Down Expand Up @@ -61,7 +62,8 @@ OpenSearch index document will look like
"{'column_name':'Letter','data_type':'string'}",
"{'column_name':'Number','data_type':'integer'}"
],
"stepId" : "s-JZSB1139WIVU"
"stepId" : "s-JZSB1139WIVU",
"applicationId" : "application_1687726870985_0003"
}
}
```
Expand Down
6 changes: 4 additions & 2 deletions spark-sql-application/src/main/scala/org/opensearch/sql/SQLJob.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,13 +84,15 @@ object SQLJob {
val schema = StructType(Seq(
StructField("result", ArrayType(StringType, containsNull = true), nullable = true),
StructField("schema", ArrayType(StringType, containsNull = true), nullable = true),
StructField("stepId", StringType, nullable = true)))
StructField("stepId", StringType, nullable = true),
StructField("applicationId", StringType, nullable = true)))

// Create the data rows
val rows = Seq((
result.toJSON.collect.toList.map(_.replaceAll("\"", "'")),
resultSchema.toJSON.collect.toList.map(_.replaceAll("\"", "'")),
sys.env.getOrElse("EMR_STEP_ID", "")))
sys.env.getOrElse("EMR_STEP_ID", "unknown"),
spark.sparkContext.applicationId))

// Create the DataFrame for data
spark.createDataFrame(rows).toDF(schema.fields.map(_.name): _*)
Expand Down
6 changes: 4 additions & 2 deletions spark-sql-application/src/test/scala/org/opensearch/sql/SQLJobTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,13 +31,15 @@ class SQLJobTest extends AnyFunSuite{
val expectedSchema = StructType(Seq(
StructField("result", ArrayType(StringType, containsNull = true), nullable = true),
StructField("schema", ArrayType(StringType, containsNull = true), nullable = true),
StructField("stepId", StringType, nullable = true)
StructField("stepId", StringType, nullable = true),
StructField("applicationId", StringType, nullable = true)
))
val expectedRows = Seq(
Row(
Array("{'Letter':'A','Number':1}","{'Letter':'B','Number':2}", "{'Letter':'C','Number':3}"),
Array("{'column_name':'Letter','data_type':'string'}", "{'column_name':'Number','data_type':'integer'}"),
""
"unknown",
spark.sparkContext.applicationId
)
)
val expected: DataFrame = spark.createDataFrame(spark.sparkContext.parallelize(expectedRows), expectedSchema)
Expand Down