wezhang

System.getProperty("sun.java.command")

wezhang

return Observable.just(appAttempt)
                 .map(attempt -> getConnectUri()
                         .resolve("/yarnui/hn/cluster/appattempt/")
                         .resolve(attempt.getAppAttemptId())
                         .toString())
                 .flatMap(this::loadPageByBrowserObservable)
                 .retry(getRetriesMax())
                 .repeatWhen(ob -> ob.delay(getDelaySeconds(), TimeUnit.SECONDS))
                 .takeUntil(this::isSparkJobYarnAppAttemptNotJustLaunched)
                 .filter(this::isSparkJobYarnAppAttemptNotJustLaunched)

wezhang

override private[scheduler] def handleJobSubmitted(jobId: Int, finalRDD: RDD[_], func: (TaskContext, Iterator[_]) => _, partitions: Array[Int], callSite: CallSite, listener: JobListener, properties: Properties): Unit = {
    val field = func.getClass.getDeclaredField("writerContainer$1")
    field.setAccessible(true)
    val writerContainer = field.get(func)
    val jobField = writerContainer.getClass.getSuperclass.getDeclaredField("job")
    jobField.setAccessible(true)
    val job = jobField.get(writerContainer).asInstanceOf[Job].getConfiguration.asInstanceOf[JobConf]
    logInfo("Output dir: " + FileOutputFormat.getOutputPath(job).toString)
    super.handleJobSubmitted(jobId, finalRDD, func, partitions, callSite, listener, properties)
  }

wezhang

    compile ('com.microsoft.azure:azure-client-runtime:1.1.1') { force = true }
    compile ('com.microsoft.azure:azure-client-authentication:1.1.1') { force = true }

wezhang

            <plugin>
                <groupId>org.codehaus.gmavenplus</groupId>
                <artifactId>gmavenplus-plugin</artifactId>
                <executions>
                    <execution>
                        <phase>generate-test-resources</phase>
                        <goals>
                            <goal>execute</goal>
                        </goals>
                    </execution>

wezhang

final Class<?> jobClass = Class.forName(jobClassName);

System.out.println("Run Spark Job: " + jobClass.getName());

final Method jobMain = jobClass.getMethod("main", String[].class);

final Object[] jobArgs = new Object[]{ args.subList(1, args.size()).toArray(new String[0]) };
jobMain.invoke(null, jobArgs);

wezhang

//@XStreamAlias("sparkJobLocalRunConfiguration")
@Tag("spark-local-run-configurable-model")
public class SparkLocalRunConfigurableModel implements CommonJavaRunConfigurationParameters {
    @XStreamAlias("isParallelExecution")
    private boolean isParallelExecution;
    @XStreamAlias("isPassParentEnvs")
    private boolean isPassParentEnvs;
    @XStreamOmitField
    @NotNull
    private Project project;

wezhang

ClassLoader.getSystemClassLoader.asInstanceOf[java.net.URLClassLoader].getURLs

wezhang

      @Mock
      def globStatus(invocation: Invocation, pathPattern: Path, filter: PathFilter): Array[FileStatus] = {

        val globberClassConstructor = Class.forName("org.apache.hadoop.fs.Globber").getDeclaredConstructor(classOf[FileSystem], classOf[Path], classOf[PathFilter])
        globberClassConstructor.setAccessible(true)

        val globber = globberClassConstructor.newInstance(invocation.getInvokedInstance, pathPattern, filter)

        Deencapsulation.invoke(globber.asInstanceOf[Object], "glob")

wezhang

        val globberClassConstructor = Class.forName("org.apache.hadoop.fs.Globber").getDeclaredConstructor(classOf[FileSystem], classOf[Path], classOf[PathFilter])
        globberClassConstructor.setAccessible(true)

        val globber = globberClassConstructor.newInstance(this, pathPattern, filter)