Я пытаюсь использовать пакет Cerner Bunsen для обработки FHIR в PySpark на AWS EMR, в частности класс Bundles и его методы. Я создаю сеанс Spark с помощью API Apache Livy,
def create_spark_session(master_dns, kind, jars):
# 8998 is the port on which the Livy server runs
host = 'http://' + master_dns + ':8998'
data = {'kind': kind, 'jars': jars}
headers = {'Content-Type': 'application/json'}
response = requests.post(host + '/sessions', data=json.dumps(data), headers=headers)
logging.info(response.json())
return response.headers
Где kind = pyspark3, а jars - это локация S3, в которой находится банка (bunsen-shaded-1.4.7.jar)
Преобразование данных пытается импортировать банку и вызывать методы через:
# Setting the Spark Session and Pulling the Existing SparkContext
sc = SparkContext.getOrCreate()
# Cerner Bunsen
from py4j.java_gateway import java_import, JavaGateway
java_import(sc._gateway.jvm,"com.cerner.bunsen.Bundles")
func = sc._gateway.jvm.Bundles()
Я получаю сообщение об ошибке
«py4j.protocol.Py4JError: ошибка при вызове None.com.cerner.bunsen.Bundles. Трассировка: \ npy4j.Py4JException: конструктор com.cerner.bunsen.Bundles ([]) не существует»
Я впервые попытался использовать java_import, поэтому я буду благодарен за любую помощь.
РЕДАКТИРОВАТЬ: Я немного изменил сценарий преобразования и теперь вижу другую ошибку. Я вижу, что jar добавляется в журналы, поэтому я уверен, что он там, и что функциональность jars: jars работает должным образом. Новое преобразование:
# Setting the Spark Session and Pulling the Existing SparkContext
sc = SparkContext.getOrCreate()
# Manage logging
#sc.setLogLevel("INFO")
# Cerner Bunsen
from py4j.java_gateway import java_import, JavaGateway
java_import(sc._gateway.jvm,"com.cerner.bunsen")
func_main = sc._gateway.jvm.Bundles
func_deep = sc._gateway.jvm.Bundles.BundleContainer
fhir_data_frame = func_deep.loadFromDirectory(spark,"s3://<bucket>/source_database/Patient",1)
fhir_data_frame_fromJson = func_deep.fromJson(fhir_data_frame)
fhir_data_frame_clean = func_main.extract_entry(spark,fhir_data_frame_fromJson,'patient')
fhir_data_frame_clean.show(20, False)
и новая ошибка:
Объект JavaPackage не вызывается
Поиск этой ошибки был немного бесполезным, но, опять же, если у кого-то есть идеи, я с радостью воспользуюсь ими.
