[BEAM-13351] FhirIO GetPatientEverything connector #15936

msbukal · 2021-11-10T18:19:52Z

Add FhirIO connector for: https://cloud.google.com/healthcare-api/docs/reference/rest/v1/projects.locations.datasets.fhirStores.fhir/Patient-everything

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

Choose reviewer(s) and mention them in a comment (R: @username).
Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
Update CHANGES.md with noteworthy changes.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

`ValidatesRunner` compliance status (on master branch)

Lang	ULR	Twister2
Go	---	---
Java
Python	---	---
XLang		---

Examples testing status on various runners

Lang	ULR	Dataflow	Flink	Samza	Spark	Twister2
Go	---	---	---	---	---	---	---
Java	---		---	---	---	---	---
Python	---	---	---	---	---	---	---
XLang	---	---	---	---	---	---	---

Post-Commit SDK/Transform Integration Tests Status (on master branch)

Go	Java	Python

Pre-Commit Tests Status (on master branch)

---	Java	Python	Go	Website	Whitespace	Typescript
Non-portable
Portable	---			---	---	---

See .test-infra/jenkins/README for trigger phrase, status and link of all Jenkins jobs.

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI.

…nto PATIENT_SOMETHING

lukecwik · 2021-11-11T19:09:53Z

R: @pabloem

msbukal · 2021-11-11T20:20:23Z

Run Java PostCommit

pabloem

looks good. Added a few comments. Aslo it seems that there's a bad merge.

pabloem · 2021-11-18T20:17:31Z

...ava/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/FhirIO.java

- * retrieve a {@link PCollection} containing the successfully fetched {@link String}s and/or {@link
- * FhirIO.Read.Result#getFailedReads()}* to retrieve a {@link PCollection} of {@link
- * HealthcareIOError}* containing the resource ID that could not be fetched and the exception as a
+ * <<<<<<< HEAD


oops there seem to be some issues with the merged result of the file

Whoops, missed these somehow because they didn't cause runtime errors haha.

pabloem · 2021-11-18T20:29:10Z

...ud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/FhirIOPatientEverything.java

+            getPatientEverything(
+                patientEverythingParameter.getResourceName(),
+                patientEverythingParameter.getFilters()));
+      } catch (IllegalArgumentException | NoSuchElementException e) {


IIUC - if there are any other exceptions, they will be thrown out right? And retried by the runner however it chooses to - correct?

Yes, that's what I believe.

pabloem · 2021-11-18T20:32:18Z

...ud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/FhirIOPatientEverything.java

+   * PatientEverythingParameter defines required attributes for a FHIR GetPatientEverything request
+   * in {@link FhirIOPatientEverything}. *
+   */
+  @DefaultCoder(PatientEverythingParameterCoder.class)


you may be able to do @DefaultSchema(JavaBeanSchema.class) for this - the advantage is that you would not have to add your own coder (though you may need to add setters and getters or constructor - or you can make it an autovalue class and use @DefaultSchema(AutoValueSchema.class)). The other advantage is that if it has a schema, it can be used easily as a cross-language transform, and users can use it with other transforms that operate on PCollections with schema.

Done, I made them AutoValue!

pabloem · 2021-11-18T20:35:27Z

...ud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/FhirIOPatientEverything.java

+    public void processElement(ProcessContext context) {
+      PatientEverythingParameter patientEverythingParameter = context.element();
+      try {
+        context.output(


if you want to parallelize the records of a single bundle, you could run something like that:

@StartBundle tpe = new ThreadPoolExecutor(); @ProcessBundle future = tpe.submit(param -> getPatientEverything(....)) futures.add(future) @FinishBundle for (f : futures) { // collect outputs or errors }

for other healthcare IO we decided to not parallelize things - and this may be a fair choice in this case - but I'm just curious if you think this may be useful

Will follow up for this.

pabloem · 2021-11-23T02:49:16Z

...ava/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/FhirIO.java

-      this.fhirStore = StaticValueProvider.of(fhirStore);
+    @Override
+    ValueProvider<String> getFhirStore() {
+      return null;


shouldn't you return the fhirStore here?

pabloem

Just added a couple questions. ready to merge after that : )

pabloem · 2021-11-23T02:50:29Z

...ava/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/FhirIO.java

+    Optional<ValueProvider<String>> getImportGcsTempPath() {
+      return Optional.empty();
+    }
+
+    @Override
+    Optional<ValueProvider<String>> getImportGcsDeadLetterPath() {
+      return Optional.empty();


do you use ValueProviders / templates in your use case? If you do not, then I think it does not make sense to add new properties that depend on value providers. Instead, just make this Optional

I'm not sure what you mean, these methods are only here because I made ExecuteBundles implement FhirIO.Write, which requires them.

pabloem · 2021-11-23T02:57:24Z

...ud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/FhirIOPatientEverything.java

+    }
+
+    @Override
+    public Map<TupleTag<?>, PValue> expand() {


are you interested in returning a Result type with getFailedReads and getPatientCompartments? This would be a little nicer to use for others, but maybe you have your reasons for going with the Map?

Sorry... isn't this what I'm doing? The return type of the FhirIOPatientEverything PTransform is Result, and there is the methods getFailedReads and getPatientCompartments, but the type POutput has to implement expand, like all the other .Result methods in FhirIO?

oh my bad! I misread this. Thanks!

pabloem · 2021-11-30T15:16:56Z

LGTM. Sorry about the delay!

y1chi · 2021-12-06T18:18:05Z

The test is failing and breaking post commit, https://issues.apache.org/jira/browse/BEAM-13387 @msbukal could you take a look at it?

msbukal · 2021-12-06T18:33:48Z

Hi, could you provide more information regarding the failed run? Is this a flake, or consistent error? Where is this test running?

The test passes locally for me. The exception is coming from line 85 but this is the exact same setup that other integration tests use (eg. FhirIOSearchIT). Probably because I run the test independently it passes, will investigate more.

y1chi · 2021-12-06T18:40:40Z

Hi, could you provide more information regarding the failed run? Is this a flake, or consistent error? Where is this test running?

The test passes locally for me. The exception is coming from line 85 but this is the exact same setup that other integration tests use (eg. FhirIOSearchIT). Probably because I run the test independently it passes, will investigate more.

You can check the jenkins job from https://issues.apache.org/jira/browse/BEAM-13387

it seems to fail consistently, but only on dataflow runner v2. On dataflow runner v1 it seems passing https://ci-beam.apache.org/job/beam_PostCommit_Java_DataflowV1/993/testReport/org.apache.beam.sdk.io.gcp.healthcare/FhirIOPatientEverythingIT/testFhirIOPatientEverything_R4_/

msbukal · 2021-12-07T16:38:08Z

How can I run the test that uses v2 dataflow runner locally?

None of the test seem to modify the bundle list, but still I think maybe changing the tests to copy the array instead of just using it might fix the issue. But I have no idea how to test this... especially since the presubmit run here passed.

y1chi · 2021-12-07T19:39:20Z

It appears that we are not running other FhirIO tests with dataflow runner v2.

beam/runners/google-cloud-dataflow-java/build.gradle

Line 557 in ed9f620

exclude '**/FhirIOReadIT.class'

I think you need either exclude the test from runner v2 if it is not intended to work on runner v2, or you can try to make

beam/runners/google-cloud-dataflow-java/build.gradle

Line 547 in ed9f620

task googleCloudPlatformRunnerV2IntegrationTest(type: Test) {

depends on copyGoogleCloudPlatformTestResources like

beam/runners/google-cloud-dataflow-java/build.gradle

Line 499 in ed9f620

    
           task googleCloudPlatformLegacyWorkerIntegrationTest(type: Test, dependsOn: copyGoogleCloudPlatformTestResources) {

(since you need to read the resource files).

You can test it locally by running the corresponding gradle task.

msbukal · 2021-12-07T19:49:44Z

Ah that makes so much sense why my new integration tests always fail.

I'll send a quick PR to disable the new test in v2 for now, and follow up to investigate enabling in v2.

msbukal added 3 commits November 10, 2021 10:19

GetPatientEverythig implementation

45fe8ae

Adding tests + comments

0c49a8f

Merge branch 'PATIENT_SOMETHING' of https://github.com/msbukal/beam i…

2f1c9a2

…nto PATIENT_SOMETHING

msbukal force-pushed the PATIENT_SOMETHING branch from adbca4d to 2f1c9a2 Compare November 11, 2021 16:54

Fix Presubmit warnings/errors

3ab9412

lukecwik requested a review from pabloem November 11, 2021 19:10

Fix checkStyle

5da5fca

pabloem reviewed Nov 18, 2021

View reviewed changes

Review fixes

95d1b3a

pabloem reviewed Nov 23, 2021

View reviewed changes

Review fixes 2

1d21e87

pabloem changed the title ~~FhirIO GetPatientEverything connector~~ [BEAM-13351] FhirIO GetPatientEverything connector Nov 30, 2021

pabloem merged commit 5ce29e7 into apache:master Nov 30, 2021

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[BEAM-13351] FhirIO GetPatientEverything connector #15936

[BEAM-13351] FhirIO GetPatientEverything connector #15936

msbukal commented Nov 10, 2021

lukecwik commented Nov 11, 2021

msbukal commented Nov 11, 2021

pabloem left a comment

pabloem Nov 18, 2021

msbukal Nov 19, 2021

pabloem Nov 18, 2021

msbukal Nov 19, 2021

pabloem Nov 18, 2021

msbukal Nov 19, 2021

pabloem Nov 18, 2021

msbukal Nov 19, 2021

pabloem Nov 23, 2021

msbukal Nov 24, 2021

pabloem left a comment

pabloem Nov 23, 2021

msbukal Nov 24, 2021

pabloem Nov 23, 2021

msbukal Nov 24, 2021

pabloem Nov 30, 2021

pabloem commented Nov 30, 2021

y1chi commented Dec 6, 2021

msbukal commented Dec 6, 2021

y1chi commented Dec 6, 2021

msbukal commented Dec 7, 2021

y1chi commented Dec 7, 2021

msbukal commented Dec 7, 2021

[BEAM-13351] FhirIO GetPatientEverything connector #15936

[BEAM-13351] FhirIO GetPatientEverything connector #15936

Conversation

msbukal commented Nov 10, 2021

ValidatesRunner compliance status (on master branch)

Examples testing status on various runners

Post-Commit SDK/Transform Integration Tests Status (on master branch)

Pre-Commit Tests Status (on master branch)

GitHub Actions Tests Status (on master branch)

lukecwik commented Nov 11, 2021

msbukal commented Nov 11, 2021

pabloem left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pabloem left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

pabloem commented Nov 30, 2021

y1chi commented Dec 6, 2021

msbukal commented Dec 6, 2021

y1chi commented Dec 6, 2021

msbukal commented Dec 7, 2021

y1chi commented Dec 7, 2021

msbukal commented Dec 7, 2021

`ValidatesRunner` compliance status (on master branch)