SPARK-REST-API

Async wrapper for Spark REST API. See more in documentation

INSTALL

pip install spark-rest-api

USAGE

import asyncio
import io
import zipfile
import aiohttp
import pandas as pd
from spark_rest_api import SparkRestApi


sra = SparkRestApi(spark_host="https://<host>:<port>")
sra.show_templates()


async def main(sra):
    base_app_id = "application_XXXXXXXXXXXXX_XXXXXX"
    async with aiohttp.ClientSession() as session:
        resp = (
            await sra.execute(
                session=session,
                url=sra.render_url(template_id=0)
            )
        )
        assert resp.status == 200
        df = resp.to_df()
        attempts = df[df["id"] == base_app_id].attempts.tolist()
        if attempts:
            urls = map(
                lambda x: sra.render_url(3, app_id=f"{base_app_id}/{x['attemptId']}"),
                attempts[0]
                )
            df_result = pd.concat(
                map(
                    lambda x: x.to_df(),
                    (
                        await asyncio.gather(*(sra.execute(session, url) for url in urls))
                        )
                    )
                )
            print(df_result)


        log_data = await sra.execute(session, sra.render_url(template_id=13, base_app_id=base_app_id))
        with zipfile.ZipFile(io.BytesIO(log_data.raw), "r") as zip_file:
            for f in zip_file.infolist():
                print(zip_file.read(f).decode())


loop = asyncio.get_event_loop()
loop.run_until_complete(main(sra))