Initial framework for pod construction architecture refactor. #1

mccheah · 2018-03-19T23:13:59Z

No description provided.

ifilonenko · 2018-03-20T04:42:43Z

...etes/core/src/main/scala/org/apache/spark/deploy/k8s/features/DriverServiceFeatureStep.scala

+    Map(DRIVER_HOST_KEY -> driverHostname,
+      "spark.driver.port" -> driverPort.toString,
+      org.apache.spark.internal.config.DRIVER_BLOCK_MANAGER_PORT.key ->
+        driverBlockManagerPort.toString)


why not driverBlockManagerPort

You have to toString here because we return Map[String, String].

ifilonenko · 2018-03-20T05:41:28Z

...urce-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/KuberneteSpec.scala

+
+import io.fabric8.kubernetes.api.model.HasMetadata
+
+private[k8s] case class KuberneteSpec(


NIT: KubernetesSpec imo

ifilonenko · 2018-03-20T19:36:32Z

...rce-managers/kubernetes/core/src/main/scala/org/apache/spark/deploy/k8s/KubernetesConf.scala

+  def sparkJars(): Seq[String] = sparkConf
+    .getOption("spark.jars")
+    .map(str => str.split(",").toSeq)
+    .getOrElse(Seq.empty[String])


additionalMainAppJar?

Plan is to have the caller inject the main app resource into the spark.jars field.

Change of plans - see KubernetesConf.createDriverConf.

Pushing through the labels from top-down instead of extracting from bottom-up. Makes KubernetesConf and KubernetesRoleSpecificConf as struct-like as possible (i.e. no logic)

mccheah · 2018-03-20T23:48:48Z

...tes/core/src/main/scala/org/apache/spark/deploy/k8s/features/KubernetesExecutorBuilder.scala

+
+import org.apache.spark.deploy.k8s.{KubernetesConf, KubernetesExecutorSpecificConf, SparkPod}
+
+private[spark] class KubernetesExecutorBuilder {


Orchestrator => Builder everywhere. Still don't know which name is better.

mccheah · 2018-03-24T02:29:53Z

When proposing against upstream, I will make the git history in such a way that each element that was moved from the old architecture to the new, will be moved in separate commits. Reviewing would most optimally be done commit by commit, but still all in the same pull request because we'd rather get the entire refactor in at once.(i.e. it's difficult to break this refactor out into logical chunks).

Almost there right now - still needs tests for KubernetesConf and each of the two builders.

mccheah · 2018-03-26T23:17:56Z

Closing in favor of apache#20910.

## What changes were proposed in this pull request? There were two related fixes regarding `from_json`, `get_json_object` and `json_tuple` ([Fix #1](apache@c8803c0), [Fix #2](apache@86174ea)), but they weren't comprehensive it seems. I wanted to extend those fixes to all the parsers, and add tests for each case. ## How was this patch tested? Regression tests Author: Burak Yavuz <[email protected]> Closes apache#20302 from brkyvz/json-invfix.

## What changes were proposed in this pull request? Solved two bugs to enable stream-stream self joins. ### Incorrect analysis due to missing MultiInstanceRelation trait Streaming leaf nodes did not extend MultiInstanceRelation, which is necessary for the catalyst analyzer to convert the self-join logical plan DAG into a tree (by creating new instances of the leaf relations). This was causing the error `Failure when resolving conflicting references in Join:` (see JIRA for details). ### Incorrect attribute rewrite when splicing batch plans in MicroBatchExecution When splicing the source's batch plan into the streaming plan (by replacing the StreamingExecutionPlan), we were rewriting the attribute reference in the streaming plan with the new attribute references from the batch plan. This was incorrectly handling the scenario when multiple StreamingExecutionRelation point to the same source, and therefore eventually point to the same batch plan returned by the source. Here is an example query, and its corresponding plan transformations. ``` val df = input.toDF val join = df.select('value % 5 as "key", 'value).join( df.select('value % 5 as "key", 'value), "key") ``` Streaming logical plan before splicing the batch plan ``` Project [key#6, value#1, value#12] +- Join Inner, (key#6 = key#9) :- Project [(value#1 % 5) AS key#6, value#1] : +- StreamingExecutionRelation Memory[#1], value#1 +- Project [(value#12 % 5) AS key#9, value#12] +- StreamingExecutionRelation Memory[#1], value#12 // two different leaves pointing to same source ``` Batch logical plan after splicing the batch plan and before rewriting ``` Project [key#6, value#1, value#12] +- Join Inner, (key#6 = key#9) :- Project [(value#1 % 5) AS key#6, value#1] : +- LocalRelation [value#66] // replaces StreamingExecutionRelation Memory[#1], value#1 +- Project [(value#12 % 5) AS key#9, value#12] +- LocalRelation [value#66] // replaces StreamingExecutionRelation Memory[#1], value#12 ``` Batch logical plan after rewriting the attributes. Specifically, for spliced, the new output attributes (value#66) replace the earlier output attributes (value#12, and value#1, one for each StreamingExecutionRelation). ``` Project [key#6, value#66, value#66] // both value#1 and value#12 replaces by value#66 +- Join Inner, (key#6 = key#9) :- Project [(value#66 % 5) AS key#6, value#66] : +- LocalRelation [value#66] +- Project [(value#66 % 5) AS key#9, value#66] +- LocalRelation [value#66] ``` This causes the optimizer to eliminate value#66 from one side of the join. ``` Project [key#6, value#66, value#66] +- Join Inner, (key#6 = key#9) :- Project [(value#66 % 5) AS key#6, value#66] : +- LocalRelation [value#66] +- Project [(value#66 % 5) AS key#9] // this does not generate value, incorrect join results +- LocalRelation [value#66] ``` **Solution**: Instead of rewriting attributes, use a Project to introduce aliases between the output attribute references and the new reference generated by the spliced plans. The analyzer and optimizer will take care of the rest. ``` Project [key#6, value#1, value#12] +- Join Inner, (key#6 = key#9) :- Project [(value#1 % 5) AS key#6, value#1] : +- Project [value#66 AS value#1] // solution: project with aliases : +- LocalRelation [value#66] +- Project [(value#12 % 5) AS key#9, value#12] +- Project [value#66 AS value#12] // solution: project with aliases +- LocalRelation [value#66] ``` ## How was this patch tested? New unit test Author: Tathagata Das <[email protected]> Closes apache#20598 from tdas/SPARK-23406.

### What changes were proposed in this pull request? Currently the join operators are not well abstracted, since there are lot of common logic. A trait can be created for easier pattern matching and other future handiness. This is a follow-up PR based on comment apache#27509 (comment) . This PR refined from the following aspects: 1. Refined structure of all physical join operators 2. Add missing joinType field for CartesianProductExec operator 3. Refined codes related to Explain Formatted The EXPLAIN FORMATTED changes are 1. Converge all join operator `verboseStringWithOperatorId` implementations to `BaseJoinExec`. Join condition displayed, and join keys displayed if it’s not empty. 2. `#1` will add Join condition to `BroadcastNestedLoopJoinExec`. 3. `#1` will **NOT** affect `CartesianProductExec`,`SortMergeJoin` and `HashJoin`s, since they already got there override implementation before. 4. Converge all join operator `simpleStringWithNodeId` to `BaseJoinExec`, which will enhance the one line description for `CartesianProductExec` with `JoinType` added. 5. Override `simpleStringWithNodeId` in `BroadcastNestedLoopJoinExec` to show `BuildSide`, which was only done for `HashJoin`s before. ### Why are the changes needed? Make the code consistent with other operators and for future handiness of join operators. ### Does this PR introduce any user-facing change? No ### How was this patch tested? Existing tests Closes apache#27595 from Eric5553/RefineJoin. Authored-by: Eric Wu <[email protected]> Signed-off-by: Wenchen Fan <[email protected]>

mccheah added 3 commits March 19, 2018 16:07

Initial framework for pod construction architecture refactor.

f480f5c

Migrate mounting credentials management to the new abstraction.

ff7e441

Move service construction to the new abstraction.

c6dbba5

ifilonenko reviewed Mar 20, 2018

View reviewed changes

mccheah added 7 commits March 20, 2018 14:17

Add mount secrets step, make role-specific confs return specific values.

ca28e8b

Simplify things by not using KubernetesRoleSpecificConf for too much.

016577b

Pushing through the labels from top-down instead of extracting from bottom-up. Makes KubernetesConf and KubernetesRoleSpecificConf as struct-like as possible (i.e. no logic)

Basic executor construction using the new abstractions.

ae2963e

editOrNewMetadata, editOrNewSpec for executors

b8a4ad8

Compose features into builders. Move private[k8s] => private[spark].

29f4c7d

Builders only selectively apply mounting secrets.

6e83749

Fix build

fe12044

mccheah commented Mar 20, 2018

View reviewed changes

mccheah added 3 commits March 20, 2018 17:36

Migrate BasicDriverConfigurationStepSuite to feature-based suite.

3c5accb

Remove BasicDriverConfigurationStep.

91450b7

Migrate a lot of tests. Remove a lot of the old code.

9d05430

mccheah added 4 commits March 26, 2018 15:00

Finish tests.

13e1f90

Move some classes around.

f21cb75

FIx some style

9ed1458

Fix style

0ec26a7

mccheah closed this Mar 26, 2018

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

Initial framework for pod construction architecture refactor. #1

Initial framework for pod construction architecture refactor. #1

Uh oh!

mccheah commented Mar 19, 2018

Uh oh!

ifilonenko Mar 20, 2018

Uh oh!

mccheah Mar 20, 2018

Uh oh!

ifilonenko Mar 20, 2018

Uh oh!

ifilonenko Mar 20, 2018

Uh oh!

mccheah Mar 20, 2018

Uh oh!

mccheah Mar 20, 2018

Uh oh!

mccheah Mar 20, 2018

Uh oh!

mccheah commented Mar 24, 2018

Uh oh!

mccheah commented Mar 26, 2018

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants


		import io.fabric8.kubernetes.api.model.HasMetadata

		private[k8s] case class KuberneteSpec(


		import org.apache.spark.deploy.k8s.{KubernetesConf, KubernetesExecutorSpecificConf, SparkPod}

		private[spark] class KubernetesExecutorBuilder {

Initial framework for pod construction architecture refactor. #1

Initial framework for pod construction architecture refactor. #1

Uh oh!

Conversation

mccheah commented Mar 19, 2018

Uh oh!

ifilonenko Mar 20, 2018

Choose a reason for hiding this comment

Uh oh!

mccheah Mar 20, 2018

Choose a reason for hiding this comment

Uh oh!

ifilonenko Mar 20, 2018

Choose a reason for hiding this comment

Uh oh!

ifilonenko Mar 20, 2018

Choose a reason for hiding this comment

Uh oh!

mccheah Mar 20, 2018

Choose a reason for hiding this comment

Uh oh!

mccheah Mar 20, 2018

Choose a reason for hiding this comment

Uh oh!

mccheah Mar 20, 2018

Choose a reason for hiding this comment

Uh oh!

mccheah commented Mar 24, 2018

Uh oh!

mccheah commented Mar 26, 2018

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants