i am trying to put 500000 records pet spark job into weaviate in each (2000 ) per batch (have a total of 8 million records ) using spark connector for weaviate. I am getting below error. Pls advise if there is a setting i need to adjust to accept this batch size on weaviate.df.write.format(“io.weaviate.spark.Weaviate”).option(“batchSize”, 2000).option(“scheme”, “http”).option(“host”
pls advise on why the above errors are happening.
2 types of issues i can see in logs -
a) {"description":"A broken pipe error occurs when Weaviate tries to write a response onto a connection that has already been closed or reset by the client. Typically, this is the case when the server was not able to respond within the configured client-side timeout.","error":"write tcp x.x.x.x:8080-\u003ex.x.x.x:36541: write: broken pipe","hint":"Either try increasing the client-side timeout, or sending a computationally cheaper request, for example by reducing a batch size, reducing a limit, using less complex filters, etc.","level":"error","method":"POST","msg":"broken pipe","path":{"Scheme":"","Opaque":"","User":null,"Host":"","Path":"/v1/batch/objects","RawPath":"","OmitHost":false,"ForceQuery":false,"RawQuery":"","Fragment":"","RawFragment":""},"time":"2023-06-21T11:06:34Z"}
b) When i set replicationfConfig factor:2. for my class then i see below error and spark job fails-
kubectl logs weaviate-0 -n sip-wvt-scaled -f | grep error
{"level":"error","msg":" memberlist: Failed fallback TCP ping: timeout 1s: read tcp x.x.x.x:36818-\u003ex.x.x.x:7000: i/o timeout","time":"2023-06-22T09:13:51Z"}
{"level":"error","msg":" memberlist: Failed fallback TCP ping: timeout 1s: read tcp x.x.x.x:33810-\u003ex.x.x.x:7000: i/o timeout","time":"2023-06-22T09:14:15Z"}
{"level":"error","msg":"\"x.x.x.x:7001\": connect: Post \"http://x.x.x.x:7001/replicas/indices/ClipEmbeddingsScaled/shards/eU2dc5Kl65uI/objects?request_id=weaviate-0-64-188e2615fd7-15\": context canceled","op":"broadcast","time":"2023-06-22T09:14:26Z"}
{"active":1,"level":"error","msg":"abort","op":"broadcast","time":"2023-06-22T09:14:26Z","total":2}
{"level":"error","msg":"\"10.249.68.130:7001\": connect: Post \"http://x.x.x.x:7001/replicas/indices/ClipEmbeddingsScaled/shards/2Mv8KFbKGP72/objects?request_id=weaviate-0-64-188e2615fd7-16\": context canceled","op":"broadcast","time":"2023-06-22T09:14:26Z"}
{"active":1,"level":"error","msg":"abort","op":"broadcast","time":"2023-06-22T09:14:26Z","total":2}
{"class":"ClipEmbeddingsScaled","level":"error","msg":"[broadcast: cannot reach enough replicas broadcast: cannot reach enough replicas broadcast: cannot reach enough replicas broadcast: cannot reach enough replicas broadcast: cannot reach enough replicas broadcast: cannot reach enough replicas broadcast: cannot reach enough replicas broadcast: cannot reach enough replicas broadcast: cannot reach enough replicas broadcast: cannot reach enough replicas broadcast: cannot reach enough rep
My vector config
class_obj = {
"class": "ClipEmbeddingsScaled",
"invertedIndexConfig": {
"indexTimestamps": True
},
"shardingConfig": {
"virtualPerPhysical": 128,
"desiredCount": 2,
"actualCount": 2,
"desiredVirtualCount": 128,
"actualVirtualCount": 128,
"key": "_id",
"strategy": "hash",
"function": "murmur3"
},
"vectorIndexConfig": {
"distance": "cosine",
"maxConnections": 250
},
"vectorizer": "multi2vec-clip",