Skip to content

Commit 5622ba7

Browse files
GedochaoGedochao
committed
Support running the test scope with --hadoop 
1 parent e9066e5 commit 5622ba7

File tree

4 files changed

+138
-127
lines changed

4 files changed

+138
-127
lines changed

‎modules/cli/src/main/scala/scala/cli/commands/package0/Package.scala‎

Lines changed: 28 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -5,6 +5,7 @@ import caseapp.*
55
import caseapp.core.help.HelpFormat
66
import coursier.launcher.*
77
import dependency.*
8+
import os.Path
89
import packager.config.*
910
import packager.deb.DebianPackage
1011
import packager.docker.DockerPackage
@@ -337,7 +338,7 @@ object Package extends ScalaCommand[PackageOptions] with BuildCommandHelpers {
337338
case a: PackageType.Assembly =>
338339
value {
339340
assembly(
340-
build,
341+
Seq(build),
341342
destPath,
342343
a.mainClassInManifest match {
343344
case None =>
@@ -367,7 +368,7 @@ object Package extends ScalaCommand[PackageOptions] with BuildCommandHelpers {
367368
case PackageType.Spark =>
368369
value {
369370
assembly(
370-
build,
371+
Seq(build),
371372
destPath,
372373
mainClassOpt,
373374
// The Spark modules are assumed to be already on the class path,
@@ -844,47 +845,52 @@ object Package extends ScalaCommand[PackageOptions] with BuildCommandHelpers {
844845
}
845846

846847
def assembly(
847-
build: Build.Successful,
848+
builds: Seq[Build.Successful],
848849
destPath: os.Path,
849850
mainClassOpt: Option[String],
850851
extraProvided: Seq[dependency.AnyModule],
851852
withPreamble: Boolean,
852853
alreadyExistsCheck: () => Either[BuildException, Unit],
853854
logger: Logger
854855
): Either[BuildException, Unit] = either {
855-
val compiledClasses = os.walk(build.output).filter(os.isFile(_))
856+
val compiledClassesByOutputDir: Seq[(Path, Path)] =
857+
builds.flatMap(build =>
858+
os.walk(build.output).filter(os.isFile(_)).map(build.output -> _)
859+
).distinct
856860
val (extraClassesFolders, extraJars) =
857-
build.options.classPathOptions.extraClassPath.partition(os.isDir(_))
858-
val extraClasses = extraClassesFolders.flatMap(os.walk(_)).filter(os.isFile(_))
859-
860-
val byteCodeZipEntries = (compiledClasses ++ extraClasses).map { path =>
861-
val name = path.relativeTo(build.output).toString
862-
val content = os.read.bytes(path)
863-
val lastModified = os.mtime(path)
864-
val ent = new ZipEntry(name)
865-
ent.setLastModifiedTime(FileTime.fromMillis(lastModified))
866-
ent.setSize(content.length)
867-
(ent, content)
868-
}
861+
builds.flatMap(_.options.classPathOptions.extraClassPath).partition(os.isDir(_))
862+
val extraClassesByDefaultOutputDir =
863+
extraClassesFolders.flatMap(os.walk(_)).filter(os.isFile(_)).map(builds.head.output -> _)
864+
865+
val byteCodeZipEntries =
866+
(compiledClassesByOutputDir ++ extraClassesByDefaultOutputDir).map { (outputDir, path) =>
867+
val name = path.relativeTo(outputDir).toString
868+
val content = os.read.bytes(path)
869+
val lastModified = os.mtime(path)
870+
val ent = new ZipEntry(name)
871+
ent.setLastModifiedTime(FileTime.fromMillis(lastModified))
872+
ent.setSize(content.length)
873+
(ent, content)
874+
}
869875

870-
val provided = build.options.notForBloopOptions.packageOptions.provided ++ extraProvided
871-
val allJars = build.artifacts.runtimeArtifacts.map(_._2) ++ extraJars.filter(os.exists(_))
876+
val provided = builds.head.options.notForBloopOptions.packageOptions.provided ++ extraProvided
877+
val allJars =
878+
builds.flatMap(_.artifacts.runtimeArtifacts.map(_._2)) ++ extraJars.filter(os.exists(_))
872879
val jars =
873880
if (provided.isEmpty) allJars
874881
else {
875-
val providedFilesSet = value(providedFiles(Seq(build), provided, logger)).toSet
882+
val providedFilesSet = value(providedFiles(builds, provided, logger)).toSet
876883
allJars.filterNot(providedFilesSet.contains)
877884
}
878885

879886
val preambleOpt =
880-
if (withPreamble)
887+
if withPreamble then
881888
Some {
882889
Preamble()
883890
.withOsKind(Properties.isWin)
884891
.callsItself(Properties.isWin)
885892
}
886-
else
887-
None
893+
else None
888894
val params = Parameters.Assembly()
889895
.withExtraZipEntries(byteCodeZipEntries)
890896
.withFiles(jars.map(_.toIO))

‎modules/cli/src/main/scala/scala/cli/commands/run/Run.scala‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -646,7 +646,7 @@ object Run extends ScalaCommand[RunOptions] with BuildCommandHelpers {
646646
case RunMode.HadoopJar =>
647647
value {
648648
RunHadoop.run(
649-
builds.head, // TODO: handle multiple builds
649+
builds,
650650
mainClass,
651651
args,
652652
logger,

‎modules/cli/src/main/scala/scala/cli/commands/util/RunHadoop.scala‎

Lines changed: 8 additions & 12 deletions
Original file line numberDiff line numberDiff line change
@@ -10,15 +10,14 @@ import scala.cli.commands.packaging.Spark
1010
object RunHadoop {
1111

1212
def run(
13-
build: Build.Successful,
13+
builds: Seq[Build.Successful],
1414
mainClass: String,
1515
args: Seq[String],
1616
logger: Logger,
1717
allowExecve: Boolean,
1818
showCommand: Boolean,
1919
scratchDirOpt: Option[os.Path]
2020
): Either[BuildException, Either[Seq[String], (Process, Option[() => Unit])]] = either {
21-
2221
// FIXME Get Spark.hadoopModules via provided settings?
2322
val providedModules = Spark.hadoopModules
2423
scratchDirOpt.foreach(os.makeDir.all(_))
@@ -30,7 +29,7 @@ object RunHadoop {
3029
)
3130
value {
3231
PackageCmd.assembly(
33-
build,
32+
builds,
3433
assembly,
3534
// "hadoop jar" doesn't accept a main class as second argument if the jar as first argument has a main class in its manifest…
3635
None,
@@ -41,27 +40,24 @@ object RunHadoop {
4140
)
4241
}
4342

44-
val javaOpts = build.options.javaOptions.javaOpts.toSeq.map(_.value.value)
43+
val javaOpts = builds.head.options.javaOptions.javaOpts.toSeq.map(_.value.value)
4544
val extraEnv =
46-
if (javaOpts.isEmpty) Map[String, String]()
45+
if javaOpts.isEmpty then Map[String, String]()
4746
else
4847
Map(
4948
"HADOOP_CLIENT_OPTS" -> javaOpts.mkString(" ") // no escaping…
5049
)
5150
val hadoopJarCommand = Seq("hadoop", "jar")
5251
val finalCommand =
5352
hadoopJarCommand ++ Seq(assembly.toString, mainClass) ++ args
54-
if (showCommand)
55-
Left(Runner.envCommand(extraEnv) ++ finalCommand)
53+
if showCommand then Left(Runner.envCommand(extraEnv) ++ finalCommand)
5654
else {
5755
val proc =
58-
if (allowExecve)
59-
Runner.maybeExec("hadoop", finalCommand, logger, extraEnv = extraEnv)
60-
else
61-
Runner.run(finalCommand, logger, extraEnv = extraEnv)
56+
if allowExecve then Runner.maybeExec("hadoop", finalCommand, logger, extraEnv = extraEnv)
57+
else Runner.run(finalCommand, logger, extraEnv = extraEnv)
6258
Right((
6359
proc,
64-
if (scratchDirOpt.isEmpty) Some(() => os.remove(assembly, checkExists = true))
60+
if scratchDirOpt.isEmpty then Some(() => os.remove(assembly, checkExists = true))
6561
else None
6662
))
6763
}

‎modules/integration/src/test/scala/scala/cli/integration/HadoopTests.scala‎

Lines changed: 101 additions & 92 deletions
Original file line numberDiff line numberDiff line change
@@ -5,99 +5,108 @@ import com.eed3si9n.expecty.Expecty.expect
55
class HadoopTests extends munit.FunSuite {
66
protected lazy val extraOptions: Seq[String] = TestUtil.extraOptions
77

8-
test("simple map-reduce") {
9-
TestUtil.retryOnCi() {
10-
val inputs = TestInputs(
11-
os.rel / "WordCount.java" ->
12-
"""//> using dep org.apache.hadoop:hadoop-client-api:3.3.3
13-
|
14-
|// from https://hadoop.apache.org/docs/r3.3.3/hadoop-mapreduce-client/hadoop-mapreduce-client-core/MapReduceTutorial.html
15-
|
16-
|package foo;
17-
|
18-
|import java.io.IOException;
19-
|import java.util.StringTokenizer;
20-
|
21-
|import org.apache.hadoop.conf.Configuration;
22-
|import org.apache.hadoop.fs.Path;
23-
|import org.apache.hadoop.io.IntWritable;
24-
|import org.apache.hadoop.io.Text;
25-
|import org.apache.hadoop.mapreduce.Job;
26-
|import org.apache.hadoop.mapreduce.Mapper;
27-
|import org.apache.hadoop.mapreduce.Reducer;
28-
|import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
29-
|import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
30-
|
31-
|public class WordCount {
32-
|
33-
| public static class TokenizerMapper
34-
| extends Mapper<Object, Text, Text, IntWritable>{
35-
|
36-
| private final static IntWritable one = new IntWritable(1);
37-
| private Text word = new Text();
38-
|
39-
| public void map(Object key, Text value, Context context
40-
| ) throws IOException, InterruptedException {
41-
| StringTokenizer itr = new StringTokenizer(value.toString());
42-
| while (itr.hasMoreTokens()) {
43-
| word.set(itr.nextToken());
44-
| context.write(word, one);
45-
| }
46-
| }
47-
| }
48-
|
49-
| public static class IntSumReducer
50-
| extends Reducer<Text,IntWritable,Text,IntWritable> {
51-
| private IntWritable result = new IntWritable();
52-
|
53-
| public void reduce(Text key, Iterable<IntWritable> values,
54-
| Context context
55-
| ) throws IOException, InterruptedException {
56-
| int sum = 0;
57-
| for (IntWritable val : values) {
58-
| sum += val.get();
59-
| }
60-
| result.set(sum);
61-
| context.write(key, result);
62-
| }
63-
| }
64-
|
65-
| public static void main(String[] args) throws Exception {
66-
| Configuration conf = new Configuration();
67-
| Job job = Job.getInstance(conf, "word count");
68-
| job.setJarByClass(WordCount.class);
69-
| job.setMapperClass(TokenizerMapper.class);
70-
| job.setCombinerClass(IntSumReducer.class);
71-
| job.setReducerClass(IntSumReducer.class);
72-
| job.setOutputKeyClass(Text.class);
73-
| job.setOutputValueClass(IntWritable.class);
74-
| FileInputFormat.addInputPath(job, new Path(args[0]));
75-
| FileOutputFormat.setOutputPath(job, new Path(args[1]));
76-
| System.exit(job.waitForCompletion(true) ? 0 : 1);
77-
| }
78-
|}
79-
|""".stripMargin
80-
)
81-
inputs.fromRoot { root =>
82-
val res = os.proc(
83-
TestUtil.cli,
84-
"--power",
85-
"run",
86-
TestUtil.extraOptions,
87-
".",
88-
"--hadoop",
89-
"--command",
90-
"--scratch-dir",
91-
"tmp",
92-
"--",
93-
"foo"
8+
for {
9+
withTestScope <- Seq(true, false)
10+
scopeDescription = if (withTestScope) "test scope" else "main scope"
11+
inputPath =
12+
if (withTestScope) os.rel / "test" / "WordCount.java" else os.rel / "main" / "WordCount.java"
13+
directiveKey = if (withTestScope) "test.dep" else "dep"
14+
scopeOptions = if (withTestScope) Seq("--test") else Nil
15+
}
16+
test(s"simple map-reduce ($scopeDescription)") {
17+
TestUtil.retryOnCi() {
18+
val inputs = TestInputs(
19+
inputPath ->
20+
s"""//> using $directiveKey org.apache.hadoop:hadoop-client-api:3.3.3
21+
|
22+
|// from https://hadoop.apache.org/docs/r3.3.3/hadoop-mapreduce-client/hadoop-mapreduce-client-core/MapReduceTutorial.html
23+
|
24+
|package foo;
25+
|
26+
|import java.io.IOException;
27+
|import java.util.StringTokenizer;
28+
|
29+
|import org.apache.hadoop.conf.Configuration;
30+
|import org.apache.hadoop.fs.Path;
31+
|import org.apache.hadoop.io.IntWritable;
32+
|import org.apache.hadoop.io.Text;
33+
|import org.apache.hadoop.mapreduce.Job;
34+
|import org.apache.hadoop.mapreduce.Mapper;
35+
|import org.apache.hadoop.mapreduce.Reducer;
36+
|import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
37+
|import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
38+
|
39+
|public class WordCount {
40+
|
41+
| public static class TokenizerMapper
42+
| extends Mapper<Object, Text, Text, IntWritable>{
43+
|
44+
| private final static IntWritable one = new IntWritable(1);
45+
| private Text word = new Text();
46+
|
47+
| public void map(Object key, Text value, Context context
48+
| ) throws IOException, InterruptedException {
49+
| StringTokenizer itr = new StringTokenizer(value.toString());
50+
| while (itr.hasMoreTokens()) {
51+
| word.set(itr.nextToken());
52+
| context.write(word, one);
53+
| }
54+
| }
55+
| }
56+
|
57+
| public static class IntSumReducer
58+
| extends Reducer<Text,IntWritable,Text,IntWritable> {
59+
| private IntWritable result = new IntWritable();
60+
|
61+
| public void reduce(Text key, Iterable<IntWritable> values,
62+
| Context context
63+
| ) throws IOException, InterruptedException {
64+
| int sum = 0;
65+
| for (IntWritable val : values) {
66+
| sum += val.get();
67+
| }
68+
| result.set(sum);
69+
| context.write(key, result);
70+
| }
71+
| }
72+
|
73+
| public static void main(String[] args) throws Exception {
74+
| Configuration conf = new Configuration();
75+
| Job job = Job.getInstance(conf, "word count");
76+
| job.setJarByClass(WordCount.class);
77+
| job.setMapperClass(TokenizerMapper.class);
78+
| job.setCombinerClass(IntSumReducer.class);
79+
| job.setReducerClass(IntSumReducer.class);
80+
| job.setOutputKeyClass(Text.class);
81+
| job.setOutputValueClass(IntWritable.class);
82+
| FileInputFormat.addInputPath(job, new Path(args[0]));
83+
| FileOutputFormat.setOutputPath(job, new Path(args[1]));
84+
| System.exit(job.waitForCompletion(true) ? 0 : 1);
85+
| }
86+
|}
87+
|""".stripMargin
9488
)
95-
.call(cwd = root)
96-
val command = res.out.lines()
97-
pprint.err.log(command)
98-
expect(command.take(2) == Seq("hadoop", "jar"))
99-
expect(command.takeRight(2) == Seq("foo.WordCount", "foo"))
89+
inputs.fromRoot { root =>
90+
val res = os.proc(
91+
TestUtil.cli,
92+
"--power",
93+
"run",
94+
TestUtil.extraOptions,
95+
".",
96+
"--hadoop",
97+
"--command",
98+
"--scratch-dir",
99+
"tmp",
100+
scopeOptions,
101+
"--",
102+
"foo"
103+
)
104+
.call(cwd = root)
105+
val command = res.out.lines()
106+
pprint.err.log(command)
107+
expect(command.take(2) == Seq("hadoop", "jar"))
108+
expect(command.takeRight(2) == Seq("foo.WordCount", "foo"))
109+
}
100110
}
101111
}
102-
}
103112
}

0 commit comments

Comments
 (0)