I have a dataframe that looks like this:
data.show()
+--------------------++-------------
| feature| id |
+--------------------++-------------
|[2.23668528E8, 1....| image1 |
|[2.23668528E8, 1....| image2 |
|[2.23668528E8, 1....| image3 |
|[2.23668528E8, 1....| image4 |
|[2.23668528E8, 1....| image5 |
I am trying to find the similarity across all the pairs. These features are basically numpy vectors extracted from vgg16 model. The number of rows are approximately 1 million and all the features are stored in s3. I am using pyspark to acheive this. I have tried the following steps:
- Fetch all the file paths from s3 in a list.
- Distribute all the paths using sc.parallelize(list of filepaths, 300) and then call a function which will download the feature from s3 in each of the executors for each of the path.
- Convert the rdd to dataframe
- Finally, call the LSH on this dataframe.
The problem is that the job keeps getting failed in this case. Meanwhile, LSH was supposed to compute similarity on millions of files in very less amount of time as suggested in these links: https://databricks.com/blog/2017/05/09/detecting-abuse-scale-locality-sensitive-hashing-uber-engineering.html.
These are spark-submit parameters I have used:
spark-submit --verbose --deploy-mode cluster --num-executors 5 --executor-cores 40 --executor-memory 110G --driver-cores 20 --driver-memory 42G --conf spark.yarn.appMasterEnv.ENVIRONMENT=dev --conf spark.executorEnv.ENVIRONMENT=dev --conf spark.yarn.appMasterEnv.REGION=us-east-1 --conf spark.executorEnv.REGION=us-east-1 --conf spark.dynamicAllocation.enabled=false --conf spark.shuffle.service.enabled=false --conf spark.dynamicAllocation.executorIdleTimeout=1200s --conf spark.network.timeout=42000s s3://<mainfile.py>
I also tried copying all the features to HDFS and following the same approach but nothing works. Any help would be greatly appreciated?