Sometimes the error is part of
"org.apache.spark.SparkException: Exception thrown in awaitResult:".
The error source is the step, where we extract the second time the rows, where the data is updated. We can count the rows, but we cannot display or write them as a table.
Our approach is:
We import our data day by day - up to 500.000 rows and, for a complete depiction of the needed information up to 105 columns, but we reduced our table already to the minimal numbers of columns of 8.
We get both new data and updates.
Instead of overwriting existing rows with our updates we would like to append them as the incremental change with respect to the existing rows.
We import one initial table.
We run the update/append algorithmus once successfully - all join, union and write commands work.
Our table schema is:
The first five columns and the last column will be called dimensions, the others are base for the increment computation :
- from which server
- the last update time (AS our import key)
- the time of import
- a classification information as integer
- the ID (AS our filter key)
- one integer type
- one double type
- the day, the row was created initially in our database (AS our partition key)
// statusIdsChanged only consists of the Id column
val statusIdsChanged = {
val statusRowsReduced = sqlContext.table(BASE_TABLE_NAME).select("Id")
statusRowsReduced.join(updatedBetId, statusRowsReduced("Id") === updatedBetId ("updateId")).drop("updateId")
}
val status = sqlContext.table(BASE_TABLE_NAME)
status.count()
def splitNamesAt(colName: org.apache.spark.sql.Dataset[Row], splitName: String) = {
val cols = colName.columns
cols.splitAt(cols.indexOf(splitName))
}
val (dimensionsStatus, incrementsStatus) = splitNamesAt(status, "Id")
// The following value is the error source.
val statusRowsChanged = status.join(statusIdsChanged, "Id")
.select(incrementsStatus.map(col):_*)
.drop("deliveryDay")
