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

Feature/1556 file access PoC using Hadoop FS API #1586

Merged
merged 12 commits into from
Nov 9, 2020
Merged

Feature/1556 file access PoC using Hadoop FS API #1586

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
12 commits
Select commit Hold shift + click to select a range
a26b4c9
s3 using hadoop fs api - first attempt to run
dk1844 Oct 26, 2020
5e7a503
s3 sdk usage removed (pom, classes, tests)
dk1844 Oct 26, 2020
17dcfbc
import cleanup, tiny updates
dk1844 Oct 26, 2020
37f440d
atum snapshot update (fs not needed for control measurement tracking,…
dk1844 Nov 2, 2020
95f4179
cleanup
dk1844 Nov 2, 2020
33c7ee2
atum final version usage
dk1844 Nov 5, 2020
cb48db7
pr touch-ups
dk1844 Nov 5, 2020
0b21ba5
FileSystems are now part of PathConfig
dk1844 Nov 6, 2020
c88bd35
PathConfigEntry -> PathWithFs
dk1844 Nov 6, 2020
768dac3
readStandardizationInputData(... path: String)(implicit ... fs: FileS…
dk1844 Nov 6, 2020
912678c
Infamous `FileSystem.toFsUtils` is now replaced by `HadoopFsUtils.get…
dk1844 Nov 6, 2020
1e36265
PR touchup: convertToSplittable's fsUtils is now a regular paramter (…
dk1844 Nov 9, 2020
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
16 changes: 7 additions & 9 deletions ...k-jobs/src/main/scala/za/co/absa/enceladus/standardization/StandardizationExecution.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,8 +57,6 @@ trait StandardizationExecution extends CommonJobExecution {
val rawFs = preparationResult.pathCfg.raw.fileSystem
val rawFsUtils = rawFs.toFsUtils

val stdFs = preparationResult.pathCfg.standardization.fileSystem

val stdDirSize = rawFsUtils.getDirectorySize(preparationResult.pathCfg.raw.path)
preparationResult.performance.startMeasurement(stdDirSize)

Expand Down Expand Up @@ -112,10 +110,9 @@ trait StandardizationExecution extends CommonJobExecution {

protected def readStandardizationInputData[T](schema: StructType,
cmd: StandardizationConfigParser[T],
path: String,
rawInput: PathWithFs,
dataset: Dataset)
(implicit spark: SparkSession,
rawFs: FileSystem,
dao: MenasDAO): DataFrame = {
val numberOfColumns = schema.fields.length
val standardizationReader = new StandardizationPropertiesProvider()
Expand All @@ -127,9 +124,9 @@ trait StandardizationExecution extends CommonJobExecution {
val inputSchema = PlainSchemaGenerator.generateInputSchema(schema, optColumnNameOfCorruptRecord)
dfReaderConfigured.schema(inputSchema)
}
val dfWithSchema = readerWithOptSchema.load(s"$path/*")
val dfWithSchema = readerWithOptSchema.load(s"${rawInput.path}/*")

ensureSplittable(dfWithSchema, path, schema)(spark, rawFs.toFsUtils)
ensureSplittable(dfWithSchema, rawInput, schema)
}

private def getColumnNameOfCorruptRecord[R](schema: StructType, cmd: StandardizationConfigParser[R])
Expand Down Expand Up @@ -222,9 +219,10 @@ trait StandardizationExecution extends CommonJobExecution {

//scalastyle:off parameter.number

private def ensureSplittable(df: DataFrame, path: String, schema: StructType)
(implicit spark: SparkSession, fsUtils: DistributedFsUtils): DataFrame = {
if (fsUtils.isNonSplittable(path)) {
private def ensureSplittable(df: DataFrame, input: PathWithFs, schema: StructType)
(implicit spark: SparkSession): DataFrame = {
implicit val fsUtils = input.fileSystem.toFsUtils
if (fsUtils.isNonSplittable(input.path)) {
convertToSplittable(df, schema)
} else {
df
Expand Down
3 changes: 1 addition & 2 deletions spark-jobs/src/main/scala/za/co/absa/enceladus/standardization/StandardizationJob.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -39,8 +39,7 @@ object StandardizationJob extends StandardizationExecution {

val preparationResult = prepareJob()
val schema = prepareStandardization(args, menasCredentials, preparationResult)
implicit val rawFs: FileSystem = preparationResult.pathCfg.raw.fileSystem
val inputData = readStandardizationInputData(schema, cmd, preparationResult.pathCfg.raw.path, preparationResult.dataset)
val inputData = readStandardizationInputData(schema, cmd, preparationResult.pathCfg.raw, preparationResult.dataset)

try {
val result = standardize(inputData, schema, cmd)
Expand Down
3 changes: 1 addition & 2 deletions ...a/za/co/absa/enceladus/standardization_conformance/StandardizationAndConformanceJob.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -36,8 +36,7 @@ object StandardizationAndConformanceJob extends StandardizationAndConformanceExe

val preparationResult = prepareJob()
val schema = prepareStandardization(args, menasCredentials, preparationResult)
implicit val rawFs = preparationResult.pathCfg.raw.fileSystem
val inputData = readStandardizationInputData(schema, cmd, preparationResult.pathCfg.raw.path, preparationResult.dataset)
val inputData = readStandardizationInputData(schema, cmd, preparationResult.pathCfg.raw, preparationResult.dataset)

try {
val standardized = standardize(inputData, schema, cmd)
Expand Down