Add examples

README.md

@@ -124,6 +124,10 @@ sink = ElasticsearchDslDatasink(index=Foo)
 
 Note that, unlike in [Elasticsearch DSL](https://elasticsearch-dsl.readthedocs.io/en/latest/), the results are not parsed as Python objects but instead remain Python dictionaries, due to Ray internally transforming everything in [Arrow format](https://arrow.apache.org/docs/python/index.html).
 
+### Examples
+
+More examples can be found in the [`examples`](examples/) directory.
+
 ## Development
 
 To build this package and contribute to its development you need to install the `build`, `setuptools` and `wheel` packages:

examples/README.md

@@ -0,0 +1,19 @@
+# Examples
+
+1. Configured your Ray cluster address in the `RAY_ADDRESS` environment variable.
+2. Adapt the runtime environment in [`env.yml`](env.yml) to include your Elasticsearch host, index, and credentials.
+3. Launch the examples like this:
+
+```shell
+# Write some test data into the index.
+ray job submit --runtime-env examples/env.yml -- python write.py
+
+# Sum the numbers of all documents in the index.
+ray job submit --runtime-env examples/env.yml -- python read.py
+
+# Sum the numbers of documents in the index that match a query.
+ray job submit --runtime-env examples/env.yml -- python read_query.py
+
+# Sum the numbers of documents in the index that match a query (using the `elasticsearch-dsl` query DSL).
+ray job submit --runtime-env examples/env.yml -- python read_query_dsl.py
+```

examples/env.yml

@@ -1,4 +1,9 @@
+working_dir: examples/
 pip:
   - elasticsearch~=8.14
   - ray-elasticsearch~=0.1.0
-
+env_vars:
+  ELASTICSEARCH_HOST: https://example.com
+  ELASTICSEARCH_INDEX: test
+  ELASTICSEARCH_USERNAME: example
+  ELASTICSEARCH_PASSWORD: example

examples/read.py

@@ -0,0 +1,21 @@
+from os import environ
+from ray import init
+from ray.data import read_datasource
+from ray_elasticsearch import ElasticsearchDatasource
+
+init()
+source = ElasticsearchDatasource(
+    index=environ["ELASTICSEARCH_INDEX"],
+    client_kwargs=dict(
+        hosts=environ["ELASTICSEARCH_HOST"],
+        http_auth=(
+            environ["ELASTICSEARCH_USERNAME"],
+            environ["ELASTICSEARCH_PASSWORD"],
+        ),
+    ),
+)
+print(f"Num rows: {source.num_rows()}")
+res = read_datasource(source)\
+    .map(lambda x: x["_source"])\
+    .sum("id")
+print(f"Read complete. Sum: {res}")

examples/read_query.py

@@ -0,0 +1,28 @@
+from os import environ
+from ray import init
+from ray.data import read_datasource
+from ray_elasticsearch import ElasticsearchDatasource
+
+init()
+source = ElasticsearchDatasource(
+    index=environ["ELASTICSEARCH_INDEX"],
+    client_kwargs=dict(
+        hosts=environ["ELASTICSEARCH_HOST"],
+        http_auth=(
+            environ["ELASTICSEARCH_USERNAME"],
+            environ["ELASTICSEARCH_PASSWORD"],
+        ),
+    ),
+    query={
+        "range": {
+            "id": {
+                "lte":  100,
+            },
+        },
+    },
+)
+print(f"Num rows: {source.num_rows()}")
+res = read_datasource(source)\
+    .map(lambda x: x["_source"])\
+    .sum("id")
+print(f"Read complete. Sum: {res}")

examples/read_query_dsl.py

@@ -0,0 +1,23 @@
+from os import environ
+from elasticsearch_dsl.query import Range
+from ray import init
+from ray.data import read_datasource
+from ray_elasticsearch import ElasticsearchDslDatasource
+
+init()
+source = ElasticsearchDslDatasource(
+    index=environ["ELASTICSEARCH_INDEX"],
+    client_kwargs=dict(
+        hosts=environ["ELASTICSEARCH_HOST"],
+        http_auth=(
+            environ["ELASTICSEARCH_USERNAME"],
+            environ["ELASTICSEARCH_PASSWORD"],
+        ),
+    ),
+    query=Range(id={"lte":  100}),
+)
+print(f"Num rows: {source.num_rows()}")
+res = read_datasource(source)\
+    .map(lambda x: x["_source"])\
+    .sum("id")
+print(f"Read complete. Sum: {res}")

examples/write.py

@@ -0,0 +1,21 @@
+from os import environ
+from ray import init
+from ray.data import range
+from ray_elasticsearch import ElasticsearchDatasink
+
+init()
+sink = ElasticsearchDatasink(
+    index=environ["ELASTICSEARCH_INDEX"],
+    client_kwargs=dict(
+        hosts=environ["ELASTICSEARCH_HOST"],
+        http_auth=(
+            environ["ELASTICSEARCH_USERNAME"],
+            environ["ELASTICSEARCH_PASSWORD"],
+        ),
+    ),
+)
+
+range(10_000)\
+    .map(lambda x: {"_source": x})\
+    .write_datasink(sink)
+print("Write complete.")