You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@iceberg.apache.org by "LucasRoesler (via GitHub)" <gi...@apache.org> on 2023/05/19 13:37:42 UTC
[GitHub] [iceberg] LucasRoesler opened a new issue, #7657: Unable to append to table after cancelled optmization procedure call
LucasRoesler opened a new issue, #7657:
URL: https://github.com/apache/iceberg/issues/7657
### Apache Iceberg version
1.2.1 (latest release)
### Query engine
Spark
### Please describe the bug 🐞
## Short version
I am unable to append data into a table after trying to optimize it.
Spark will attempt to load and sort the entire table before inserting the new batch even though we are partitioning by `loaded_at` which should never need more than one partition.
I am not sure how to correctly investigate the table and possibly fix it. Are there any written recommendations for how to verify the table metadata or to figure out why Spark is trying to sort the entire table?
## Long Version
We have a Spark job that uses Spark Structured Streaming to load data from Kafka into a an Iceberg table called `loading_zone`. This Kafka topic receives data on a nearly continuous basis. The spark job does some light parsing and anonymization of the data and then writes the data to the `loading_zone` table which is partitioned by a `loaded_at` timestamp, using an hourly transform. As of this writing, the table was approximately 2.8 billion records.
We are using AWS Glue for the catalog and S3 for storage, so that we can also query the data from Athena.
Last week I tried to follow some of the recommendations about table maintenance here
* https://iceberg.apache.org/docs/latest/maintenance/
* https://iceberg.apache.org/docs/latest/spark-procedures/
For various reasons one of the procedures was cancelled and we thought that the table was still in a good state. Unfortunately, we didn't immediately detect the issue and I don't know exactly which procedure was cancelled, but I believe it was the `rewrite_data_files` procedure.
On Monday, I checked the status of the table and noticed that it had not loaded any data over the weekend.
I then started checking the pod logs for the Spark job and the Spark UI to determine what was happening. Two things immediately stood out:
* The Spark UI reported that it was processing 0 rows per second.
* The Pod logs were very weird, it was continuously updating the offsets for the kafka consumer, every partition of the kafka topic was being updated several times per second. All of the logs came from `org.apache.kafka.clients.consumer.internals.SubscriptionState` For example:
> "Seeking to EARLIEST offset of partition mpathic-event-18"
> "Seeking to EARLIEST offset of partition mpathic-event-23"
> "Resetting offset for partition mpathic-event-16 to position FetchPosition{offset=183777608, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=Optional[contextstore-kafka-1.contextstore-kafka-headless.customer-insights.svc.cluster.local:9092 (id: 1 rack: null)], epoch=125}}."
> "Seeking to LATEST offset of partition mpathic-event-16"
Seeking is not unusual, but at that speed makes no sense when we have a 1 minute trigger for the spark stream.
At first we thought we had a issue in the Kafka cluster, but we were able to rule that out because there are several other consumers that did not have any issues with the same data. We could also rule out any additional deployments or configuration changes in Kafka and we could also rule out unusual data flow, in fact the total data and the rate of data in the topic during the weekend was lower than earlier in the week when the job was working fine. We decided Kafka was not the source of the problem.
Eventually we started looking into the Spark UI to get more information about the job. We noticed that
1. the job was using a lot of disk space and essentially no memory
2. the batch duration was extremely long, the entire lifetime of the job
3. kubernetes would eventually kill it because of the high disk usage after approximately 90 mins (sometimes longer, sometimes shorter)
We tried reducing the size of the batches, but nothing helped.
Finally we looked at the DataFrame metrics and planning and noticed that it was scanning _a lot_ of data. 69 million rows and growing, which made no sense because we had made the batch size only a couple thousand kafka records.
We looked at the query plan and it said it was doing a Sort
```
== Physical Plan ==
AppendData (5)
+- Sort (4)
+- Exchange (3)
+- Project (2)
+- * Scan ExistingRDD (1)
```
This is also not unusual because data needs to be sorted within the partition. Finally we decided/realized that it must be trying to sort the entire table, not just the partition.
In the end we changed the Spark job to write to a new table so that we didn't lose anything from Kafka. Once we changed to a new table, the job started working again.
I would like to recover and merge these two tables together but I am not sure how to really debug the table or it's metadata. I don't understand why the spark job changed it's behavior and needed to scan/sort the entire table.
I also have questions about the best (most efficient) way to merge the tables together. Should i just use something like
```sql
REPLACE TABLE loading_zone
USING iceberg
PARTITIONED BY (hours(loaded_at))
TBLPROPERTIES (
"format-version" = "2"
)
AS SELECT * from loading_zone UNION ALL SELECT * from loading_zone_b
```
or would something like [`add_files`](https://iceberg.apache.org/docs/latest/spark-procedures/#add_files) be better?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] LucasRoesler commented on issue #7657: Unable to append to table after cancelled optmization procedure call
Posted by "LucasRoesler (via GitHub)" <gi...@apache.org>.
LucasRoesler commented on issue #7657:
URL: https://github.com/apache/iceberg/issues/7657#issuecomment-1554792856
Sorry about not including the code snippet earlier. We are using PySpark.
We are using `foreachBatch` and then just appending the batch dataframe to the table like this
```py
df.writeTo(table_identifier)
.using("iceberg")
.option("fanout-enabled", True)
.append()
```
The full loader code is here, lightly modified to remove some comments and in-lining a few config values, etc:
<details>
```py
LOADING_ZONE_TABLE = "loading_zone_b"
def process_batch(spark: SparkSession, opts: Options):
def process(df: DataFrame, id: int):
if df.isEmpty():
return
df = df.withColumns(
{
"loaded_at": F.lit(F.current_timestamp()),
"loader_metadata": F.struct(
F.lit(__version__).alias("version"),
F.lit(__tag__).alias("tag"),
F.lit(loader_additional).alias("additional"),
),
}
)
df = df.drop("processed").withColumnRenamed("processing_error", "loader_error")
cfg = opts.iceberg
table_identifier = cfg.table_identifier(LOADING_ZONE_TABLE)
(
df.writeTo(table_identifier)
.using("iceberg")
.option("fanout-enabled", True)
.append()
)
return process
def loader(spark: SparkSession, opts: Options):
df = (
spark.readStream.format("kafka")
.options(**opts.kafka.to_dict())
.option("subscribe", opts.kafka.topic)
.load()
)
df = df.withColumn(
"topic",
F.col("topic").cast("string")
).withColumn(
"value",
F.col("value").cast("string")
)
stream = (
df.select(
F.col("topic").alias("kafka_topic"),
F.col("timestamp").alias("kafka_received_at"),
F.col("timestamp").alias("kafka_timestamp"),
parser(
F.lit(opts.hashing.id_hash_password.get_secret_value()),
F.col("value"),
).alias("event_data"),
)
.select(
F.col("kafka_topic"),
F.col("kafka_received_at"),
F.col("event_data.*"),
)
.writeStream.queryName("loader")
.trigger(processingTime=opts.iceberg.processing_time)
.option("checkpointLocation", opts.kafka.checkpointLocation)
.foreachBatch(process_batch(spark, opts))
.start()
)
stream.awaitTermination()
```
</details>
You mention that the plan should give details about where the ExistingRDD is coming from but since I am still somewhat new to Spark, I just don't realize it when I am looking at it
```txt
== Physical Plan ==
AppendData (5)
+- Sort (4)
+- Exchange (3)
+- Project (2)
+- * Scan ExistingRDD (1)
(1) Scan ExistingRDD [codegen id : 1]
Output [17]: [kafka_topic#440073, kafka_received_at#440074, processed#440075, processing_error#440076, source#440077, source_event_id#440078, recorded_at#440079, recorded_by#440080, token#440081, event_type#440082, event_name#440083, event_type_version#440084, person_id#440085, ids#440086, collector#440087, event#440088, raw#440089]
Arguments: [kafka_topic#440073, kafka_received_at#440074, processed#440075, processing_error#440076, source#440077, source_event_id#440078, recorded_at#440079, recorded_by#440080, token#440081, event_type#440082, event_name#440083, event_type_version#440084, person_id#440085, ids#440086, collector#440087, event#440088, raw#440089], SQLExecutionRDD[35818] at start at <unknown>:0, ExistingRDD, UnknownPartitioning(0)
(2) Project
Output [18]: [kafka_topic#440073, kafka_received_at#440074, source#440077, source_event_id#440078, recorded_at#440079, recorded_by#440080, token#440081, event_type#440082, event_name#440083, event_type_version#440084, person_id#440085, transform(ids#440086, lambdafunction(if (isnull(lambda x#440204)) null else named_struct(id, lambda x#440204.id, id_type, lambda x#440204.id_type), lambda x#440204, false)) AS ids#440211, if (isnull(collector#440087)) null else named_struct(id, collector#440087.id, received_at, collector#440087.received_at, hash, collector#440087.hash) AS collector#440218, event#440088, raw#440089, 2023-05-19 15:53:01.001032 AS loaded_at#440110, processing_error#440076 AS loader_error#440149, [3.18.4,v3.18.4,{}] AS loader_metadata#440225]
Input [17]: [kafka_topic#440073, kafka_received_at#440074, processed#440075, processing_error#440076, source#440077, source_event_id#440078, recorded_at#440079, recorded_by#440080, token#440081, event_type#440082, event_name#440083, event_type_version#440084, person_id#440085, ids#440086, collector#440087, event#440088, raw#440089]
(3) Exchange
Input [18]: [kafka_topic#440073, kafka_received_at#440074, source#440077, source_event_id#440078, recorded_at#440079, recorded_by#440080, token#440081, event_type#440082, event_name#440083, event_type_version#440084, person_id#440085, ids#440211, collector#440218, event#440088, raw#440089, loaded_at#440110, loader_error#440149, loader_metadata#440225]
Arguments: hashpartitioning(iceberghourtransform(loaded_at#440110), 200), REPARTITION_BY_NUM, [plan_id=199587]
(4) Sort
Input [18]: [kafka_topic#440073, kafka_received_at#440074, source#440077, source_event_id#440078, recorded_at#440079, recorded_by#440080, token#440081, event_type#440082, event_name#440083, event_type_version#440084, person_id#440085, ids#440211, collector#440218, event#440088, raw#440089, loaded_at#440110, loader_error#440149, loader_metadata#440225]
Arguments: [iceberghourtransform(loaded_at#440110) ASC NULLS FIRST], false, 0
(5) AppendData
Input [18]: [kafka_topic#440073, kafka_received_at#440074, source#440077, source_event_id#440078, recorded_at#440079, recorded_by#440080, token#440081, event_type#440082, event_name#440083, event_type_version#440084, person_id#440085, ids#440211, collector#440218, event#440088, raw#440089, loaded_at#440110, loader_error#440149, loader_metadata#440225]
Arguments: org.apache.spark.sql.execution.datasources.v2.DataSourceV2Strategy$$Lambda$5479/0x0000000841df1c40@a4ed326, IcebergWrite(table=glue.dwh.loading_zone_b, format=PARQUET)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] LucasRoesler commented on issue #7657: Unable to append to table after cancelled optmization procedure call
Posted by "LucasRoesler (via GitHub)" <gi...@apache.org>.
LucasRoesler commented on issue #7657:
URL: https://github.com/apache/iceberg/issues/7657#issuecomment-1554741515
We suspected the `rewrite-datafiles` because it is the only change that we know happened around the time that we stopped loading data.
Regarding
> If you are confident it does not need to be shuffled you can always set the mode to none.
We are currently just grabbing a batch of message from kafka and then we set the loaded time to the current timestamp. I suspect we don't need a suffle here, all of the data should just go to the end of the current partition. I am not 100% sure though,
Regarding the version, we did update the iceberg version while testing various fixes for this, from 1.2.0 to 1.2.1. But it had been running 1.2.0 for a couple weeks or longer. If I read the changelog correctly, 1.2.0 already had this change to the `write.distribution-mode`.
Part of the reason we suspected that it was trying to sort the full table is because it is the only thing that has as much data as it was claiming to have loaded, at one point it had sorted through 69 million rows. The kafka topic simply didn't have that much data in it for the time period specified
For example, this is one of the screenshots I can find in our slack
<details>
</details>
Compared to the same screen for the currently running stream, i can't find any that show more rows loaded
<details>
</details>
Additionally, a single partition doesn't really have enough data to account for the same of that `ExistingRDD`
This is the metadata for the last handful of partitions in the new table
<details>
</details>
For comparison, the same partition data for the original table
<details>
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [I] Unable to append to table after cancelled optmization procedure call [iceberg]
Posted by "LucasRoesler (via GitHub)" <gi...@apache.org>.
LucasRoesler closed issue #7657: Unable to append to table after cancelled optmization procedure call
URL: https://github.com/apache/iceberg/issues/7657
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] LucasRoesler commented on issue #7657: Unable to append to table after cancelled optmization procedure call
Posted by "LucasRoesler (via GitHub)" <gi...@apache.org>.
LucasRoesler commented on issue #7657:
URL: https://github.com/apache/iceberg/issues/7657#issuecomment-1554812215
It is most likely that my assessment and initial ideas about the issue are wrong, but the only thing I changed to fix the loader was changing the target to a new empty table, it started working immediately. None of kafka or stream configuration changes had any impact.
If there are other commands or configs I should adjust and then report back, it is still possible for me to revert back to the previous configuration and investigate it continuing the old stream into the old table. I just don't know where to start in terms of next steps.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on issue #7657: Unable to append to table after cancelled optmization procedure call
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on issue #7657:
URL: https://github.com/apache/iceberg/issues/7657#issuecomment-1554800037
Looks like a standard read from Kafka there, shuffle is there because of the write hash distribution. Nothing in that plan about reading from an Iceberg relation. Although there might be something in the Stream Code i'm missing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on issue #7657: Unable to append to table after cancelled optmization procedure call
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on issue #7657:
URL: https://github.com/apache/iceberg/issues/7657#issuecomment-1554757626
You can just look at the plan details to see what "existingRDD" is being generated from. Although this should be in your code. Again, and I can't emphasize this enough, there is no way an "Insert" can read the data already in your table. Only an Update, Merge or Delete can do that.
You really should share the actual code you are using (or sql) for insert because comparing things based on their approximate size is probably not going to get us very far.
And again on the coincidence of rewrite_data_files, the operation can only do 1 thing to modify the table for future readers, and that is to change the metadata.json pointed to in the catalog. All Iceberg transactions function on this basic principal. If that pointer did not change, there is no way the command can influence future readers. The command may leave garbage files behind though if it failed, hence my thoughts about resource issues. Any of the other things I could imagine, like removing existing metadata or data files accidentally (which rewrite data files does not do but there are mysteries everywhere) would result in abrupt errors.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
[GitHub] [iceberg] RussellSpitzer commented on issue #7657: Unable to append to table after cancelled optmization procedure call
Posted by "RussellSpitzer (via GitHub)" <gi...@apache.org>.
RussellSpitzer commented on issue #7657:
URL: https://github.com/apache/iceberg/issues/7657#issuecomment-1554603272
The `rewrite-datafiles` is almost certainly a red herring. It shouldn't have any effect on the table if it failed since it is transactional like all Iceberg operations. It may be that this is coincidental or perhaps the system has just run out of some resource like disk space.
> Spark will attempt to load and sort the entire table before inserting the new batch even though we are partitioning by loaded_at which should never need more than one partition.
This is quite confusing to me. An "append" should never require reading anything from the target table. The source though will have a shuffle applied to it based on the `write.distribution-mode` of the table. We recently did change the default to hash which would request that the incoming data be shuffled by the partition of the destination table. If you are confident it does not need to be shuffled you can always set the mode to `none`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [I] Unable to append to table after cancelled optmization procedure call [iceberg]
Posted by "LucasRoesler (via GitHub)" <gi...@apache.org>.
LucasRoesler commented on issue #7657:
URL: https://github.com/apache/iceberg/issues/7657#issuecomment-1813875337
I haven't been able to reproduce this since it happened. I will close it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org
Re: [I] Unable to append to table after cancelled optmization procedure call [iceberg]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on issue #7657:
URL: https://github.com/apache/iceberg/issues/7657#issuecomment-1813498512
This issue has been automatically marked as stale because it has been open for 180 days with no activity. It will be closed in next 14 days if no further activity occurs. To permanently prevent this issue from being considered stale, add the label 'not-stale', but commenting on the issue is preferred when possible.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@iceberg.apache.org
For additional commands, e-mail: issues-help@iceberg.apache.org