# AWS, es-hadoop and 429 **URL:** \ **Category:** Elasticsearch\ **Tags:** es-hadoop\ **Created:** [July 20, 2022, 9:37am UTC](https://discuss.elastic.co/t/aws-es-hadoop-and-429/310124 "2022-07-20T09:37:22Z")\ **Posts on this page:** 7\ **Page:** 1
**Author:** ![matze79](https://avatars.discourse-cdn.com/v4/letter/m/f08c70/32.png) [@matze79](https://discuss.elastic.co/u/matze79)\ **Post date:** [July 20, 2022, 9:37am UTC](https://discuss.elastic.co/t/aws-es-hadoop-and-429/310124/1 "2022-07-20T09:37:22Z")
Hi forum, I'm on AWS and trying to write ~ 1.2mio documents from an AWS Glue job Python / pyspark job to a managed ElasticSearch. To be fair mentioning that I'm using AWS OpenSearch 1.2 but my question is more or less about Elastic-Hadoop integration at all. The issue I'm facing is that after a while I'm facing "429 Too Many Requests": org.elasticsearch.hadoop.rest.EsHadoopInvalidRequest: [PUT] on [MY/doc/\_bulk] failed; server[https://SOME\_ENDPOINT\_HERE] returned [429|Too Many Requests:] From what I understand and read so far this is pretty much about configuration, throttling down indexing requests on the client side giving the server more time to process queued requests. And that's what I tried but somehow the config on the Hadoop connector does not work for me. So I tried to send smaller batches of documents to the ElasticSearch and increased retry wait time: set 'es.batch.size.entries' to 100 and 'es.batch.write.retry.wait' to 30s: ```auto df \ .write \ .mode('overwrite') \ .format('org.elasticsearch.spark.sql') \ .option('es.nodes', 'SOME_ENDPOINT_HERE') \ .option('es.port', 443) \ .option('es.net.ssl', 'true') \ .option('es.net.http.auth.user', 'SOME_USER_NAME_HERE') \ .option('es.net.http.auth.pass', 'SOME_PASS_HERE') \ .option('es.nodes.wan.only', 'true') \ .option('es.nodes.discovery', 'false') \ .option('es.resource', 'SOME_NAME_HERE') \ .option('es.index.auto.create', 'true') \ .option('es.mapping.id', 'SOME_FIELD_HERE') \ .option('es.write.operation', 'index') \ .option('es.batch.size.entries', '100') \ .option('es.batch.write.retry.policy', 'simple') \ .option('es.batch.write.retry.count', '-1') \ .option('es.batch.write.retry.limit', '-1') \ .option('es.batch.write.retry.wait', '30s') \ .save() ``` Already set logging for 'org.elasticsearch.hadoop.rest' logger to DEBUG level: ```auto Bulk Flush #[12653715211658247214404]: Sending batch of [34000] bytes/[1000] entries Bulk Flush #[12653715211658247214404]: Response received Bulk Flush #[12653715211658247214404]: Completed. [1000] Original Entries. [1] Attempts. [1000/1000] Docs Sent. [0/1000] Docs Skipped. [0/1000] Docs Aborted. ``` From what I understand the Hadoop-Connector is sending batches of 1000 documents, not the 100 from my config. Further I can not see any wait time. My actual setup on AWS is: Spark: 2.4.3 Python: 3.7 Elasticsearch: 7.10 / OpenSearch 1.2 Elasticsearch Hadoop: 7.13.4 Any hints or ideas on my setup? Many Thanks, Matthias ---
**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\ **Post date:** [July 20, 2022, 9:37am UTC](https://discuss.elastic.co/t/aws-es-hadoop-and-429/310124/2 "2022-07-20T09:37:23Z")
OpenSearch/OpenDistro are AWS run products and differ from the original Elasticsearch and Kibana products that Elastic builds and maintains. You may need to contact them directly for further assistance. (This is an automated response from your friendly Elastic bot. Please report this post if you have any suggestions or concerns :elasticheart: ) ---
**Author:** ![Keith\_Massey](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/keith_massey/32/83666_2.png) [@Keith\_Massey](https://discuss.elastic.co/u/Keith_Massey)\ **Post date:** [July 20, 2022, 5:17pm UTC](https://discuss.elastic.co/t/aws-es-hadoop-and-429/310124/3 "2022-07-20T17:17:38Z")
This looks like it might be a bug. I tried it in both scala and python (in a docker container running spark in yarn), and saw the same thing you did. Interestingly (and frustratingly), it works fine if I add this as an integration test in the code base. For reference, here's the python code I tried that failed: ```auto data = [("Java", "20000"), ("Python", "100000"), ("Scala", "3000")] rdd = sc.parallelize(data) df = rdd.toDF() sc.setLogLevel('ALL') df.write.mode('overwrite').format('org.elasticsearch.spark.sql').option('es.batch.size.entries', 1).option('es.resource', 'myindex').save() sc.setLogLevel('INFO') ``` And here's the analogous scala code that failed: ```auto val data = Seq(("Java", 20000), ("Python", 100000), ("Scala", 3000)) val rdd = sc.parallelize(data) val df = rdd.toDF() sc.setLogLevel("ALL") df.write.mode("overwrite").format("org.elasticsearch.spark.sql").option("es.batch.size.entries", 1).option("es.resource", "myindex").save() sc.setLogLevel("INFO") ``` In both cases, if I look for "Sending batch" in the logs, I see that it sent all 3 instead of 1. ---
**Author:** ![Keith\_Massey](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/keith_massey/32/83666_2.png) [@Keith\_Massey](https://discuss.elastic.co/u/Keith_Massey)\ **Post date:** [July 20, 2022, 6:00pm UTC](https://discuss.elastic.co/t/aws-es-hadoop-and-429/310124/4 "2022-07-20T18:00:24Z")
Oh, as a coworker just pointed out, the problem is here: [elasticsearch-hadoop/DefaultSource.scala at main · elastic/elasticsearch-hadoop · GitHub](https://github.com/elastic/elasticsearch-hadoop/blob/main/spark/sql-30/src/main/scala/org/elasticsearch/spark/sql/DefaultSource.scala#L621). In overwrite mode, first the data is deleted using a hard-coded `es.batch.size.entries` of 1000. That's what both of us were seeing in the driver log (I'm assuming). The actual writes are happening in the tasks, using the `es.batch.size.entries` that you specified. Unfortunately `es.batch.size.entries` is not configurable for that initial delete. You can either delete the documents ahead of time yourself, or avoid using overwrite. We probably ought to make the delete respect `es.batch.size.entries` if you set it, but that's not how it works right now. ---
**Author:** ![warkolm](https://sea2.discourse-cdn.com/elastic/user_avatar/discuss.elastic.co/warkolm/32/39224_2.png) [@warkolm](https://discuss.elastic.co/u/warkolm)\ **Post date:** [July 20, 2022, 9:41pm UTC](https://discuss.elastic.co/t/aws-es-hadoop-and-429/310124/5 "2022-07-20T21:41:26Z")
Please also note however that Elasticsearch 7.10 is [EOL](https://www.elastic.co/support/eol) and no longer supported. ---
**Author:** ![matze79](https://avatars.discourse-cdn.com/v4/letter/m/f08c70/32.png) [@matze79](https://discuss.elastic.co/u/matze79)\ **Post date:** [July 21, 2022, 12:18pm UTC](https://discuss.elastic.co/t/aws-es-hadoop-and-429/310124/7 "2022-07-21T12:18:51Z")
Hi, thanks for your quick reply @Keith and @Mark. Nice catch with "if (overwrite)" and the hard-coded batch size. In case this is of interest to someone else: Cleaned up the index manually beforehand and also adjusted the AWS instance sizing to "medium" (suggested by AWS support folks). The indexing of the 1.2mio documents is now successfully completed in ~7min, even with es.batch.size.entries=1000: ```auto df \ .write\ .mode('overwrite') \ ... .option('es.batch.size.entries', '1000') \ .option('es.batch.size.bytes', '250000') \ .option('es.batch.write.retry.policy', 'simple') \ .option('es.batch.write.retry.count', '-1') \ .option('es.batch.write.retry.limit', '-1') \ .option('es.batch.write.retry.wait', '30s') \ .save() ``` Again thanks for the support. Best, Matthias ---
**Author:** ![system](https://us1.discourse-cdn.com/elastic/original/3X/1/a/1ac57faf039f6b580b3f104ef42a2a89e41014de.png) [@system](https://discuss.elastic.co/u/system)\ **Post date:** [August 18, 2022, 12:19pm UTC](https://discuss.elastic.co/t/aws-es-hadoop-and-429/310124/8 "2022-08-18T12:19:23Z")
This topic was automatically closed 28 days after the last reply. New replies are no longer allowed.