Access to external Elastic Search index with PySpark code

Ask the Community
,
1

Hello,

I’m working on use case to access a on premise Elastic Search index from my Foundry stack.

For it, with Egress/Generic data source, I’m able to access it from a Python Code Repository.

But it works only in Python (i.e. using Elasticsearch library) and not in PySpark.

Indeed, when I want to read/write in this index, we PySpark, I have this error:

transforms.external.systems._redact_credentials_in_output.Py4JJavaError: An error occurred while calling o362.load.
: org.apache.spark.SparkClassNotFoundException: [DATA_SOURCE_NOT_FOUND] Failed to find the data source: org.<redacted>search.spark.sql. Please find packages at `https://spark.apache.org/third-party-projects.html`.
	at org.apache.spark.sql.errors.QueryExecutionErrors$.dataSourceNotFoundError(QueryExecutionErrors.scala:725)

Below, the code I use for it:

@external_systems(elastic_search_source=Source("xxxxx"))
@transform(
    output=Output("yyyyy"),
    source=Input("zzzzz"),
)
def compute(ctx, source, output, elastic_search_source):

    user = elastic_search_source.get_secret("User")
    password = elastic_search_source.get_secret("Password")

    df = source.dataframe()

    es_options = {
        "es.nodes": "xxx",
        "es.port": "9200",
        "es.resource": "test",
        "es.nodes.wan.only": "true",
        "es.net.http.auth.user": user,
        "es.net.http.auth.pass": password,
        "es.net.ssl": "true",
        "es.net.ssl.cert.allow.self.signed": "true",
        "es.mapping.id": "id",
        "es.batch.size.entries": "1000",
        "es.batch.size.bytes": "5mb",
        "es.batch.write.retry.count": "3",
        "es.batch.write.retry.wait": "10s",
        "es.index.auto.create": "true"
    }

    df = ctx.spark_session.read.format("org.elasticsearch.spark.sql").options(**es_options).load("test")

Would you know how I could access this Elastic Search with PySpark code ?

Thanks.

Xavier