ReservedConcurrentExecutions Removed

export-neptune-to-elasticsearch/cloudformation-templates/export-neptune-to-elasticsearch.json

@@ -551,8 +551,7 @@
         },
         "MemorySize": 128,
         "Runtime": "python3.9",
-        "Timeout": "30",
-        "ReservedConcurrentExecutions": 1
+        "Timeout": "30"
       }
     },
     "KinesisToElasticSearchLambda": {
@@ -654,15 +653,15 @@
             "STREAM_NAME": {
               "Ref": "KinesisStream"
             },
+            "ELASTIC_SEARCH_ENDPOINT": {
+              "Ref": "ElasticSearchEndpoint"
+            },
             "log_commit_nums": "false"
           }
         },
         "MemorySize": 1024,
         "Runtime": "python3.9",
         "Timeout": "900",
-        "ReservedConcurrentExecutions": {
-          "Ref": "KinesisShardCount"
-        },
         "VpcConfig": {
           "SecurityGroupIds": [
             {

export-neptune-to-elasticsearch/lambda/export_neptune_to_kinesis.py

@@ -16,14 +16,61 @@
 import boto3
 import logging
 from datetime import datetime
+from opensearchpy import OpenSearch, RequestsHttpConnection, AWSV4SignerAuth
 
 client = boto3.client('batch')
 
 logger = logging.getLogger()
 logger.setLevel(logging.INFO)
 
+def pre_trigger():
+    # clears `amazon_neptune` indices before we run rest of the job
+    print("Started PreTrigger Job")
+    host = os.environ['ELASTIC_SEARCH_ENDPOINT'] # cluster endpoint, for example: my-test-domain.us-east-1.es.amazonaws.com
+    region = os.environ['AWS_REGION']
+    service = 'es'
+    credentials = boto3.Session().get_credentials()
+    auth = AWSV4SignerAuth(credentials, region, service)
+
+    client = OpenSearch(
+        hosts = [{'host': host, 'port': 443}],
+        http_auth = auth,
+        use_ssl = True,
+        verify_certs = True,
+        connection_class = RequestsHttpConnection,
+        pool_maxsize = 20
+    )
+
+    logger.info("Listing Indices Now")
+    response = client.indices.get_alias("*")
+    logger.info(response)
+
+    logger.info("Creating Index Now")
+    index_name = "amazon_neptune"
+    index_body = {
+        "settings": {
+            "number_of_shards": 3,
+            "number_of_replicas": 2,
+        },
+    }
+    response = client.indices.create(index=index_name, body=index_body)
+    logger.info(response)
+
+    logger.info("Listing Indices Now")
+    response = client.indices.get_alias("*")
+    logger.info(response)
+
+    logger.info("Deleting Index Now")    
+    response = client.indices.delete(index="your_index_name")
+    logger.info(response)
+
+    logger.info("Listing Indices Now")
+    response = client.indices.get_alias("*")
+    logger.info(response)
+
 def trigger_neptune_export():
 
+    pre_trigger()
     neptune_export_jar_uri = os.environ['NEPTUNE_EXPORT_JAR_URI']
     neptune_endpoint = os.environ['NEPTUNE_ENDPOINT']
     neptune_port = os.environ['NEPTUNE_PORT']

export-neptune-to-elasticsearch/lambda/requirements.txt

@@ -1,10 +1,23 @@
-protobuf==3.18.3
-six==1.16.0
+aws-kinesis-agg==1.1.7
+cachetools==5.3.1
+certifi==2023.5.7
+charset-normalizer==3.1.0
+distlib==0.3.6
 elasticsearch==6.4.0
+filelock==3.12.2
+idna==3.4
+isodate==0.6.1
+jmespath==1.0.1
+opensearch-py==2.2.0
+platformdirs==3.7.0
+protobuf==3.18.3
+pyparsing==3.1.0
+python-dateutil==2.8.2
 rdflib==6.2.0
-retrying
-requests
-requests_aws4auth
-cachetools
-certifi
-aws-kinesis-agg==1.1.7
+requests==2.31.0
+requests-aws4auth==1.2.3
+retrying==1.3.4
+s3transfer==0.6.1
+six==1.16.0
+urllib3==1.26.16
+virtualenv==20.23.1