You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2021/11/10 18:20:09 UTC
[GitHub] [beam] msbukal opened a new pull request #15936: FhirIO GetPatientEverything connector
msbukal opened a new pull request #15936:
URL: https://github.com/apache/beam/pull/15936
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)**](https://beam.apache.org/contribute/#make-your-change) 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](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://beam.apache.org/contribute/#make-reviewers-job-easier).
`ValidatesRunner` compliance status (on master branch)
--------------------------------------------------------
<table>
<thead>
<tr>
<th>Lang</th>
<th>ULR</th>
<th>Dataflow</th>
<th>Flink</th>
<th>Samza</th>
<th>Spark</th>
<th>Twister2</th>
</tr>
</thead>
<tbody>
<tr>
<td>Go</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/badge/icon">
</a>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Samza/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Samza/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
</tr>
<tr>
<td>Java</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_ULR/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_ULR/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/badge/icon?subject=V1">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Streaming/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Streaming/lastCompletedBuild/badge/icon?subject=V1+Streaming">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/badge/icon?subject=V1+Java+11">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2/lastCompletedBuild/badge/icon?subject=V2">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2_Streaming/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2_Streaming/lastCompletedBuild/badge/icon?subject=V2+Streaming">
</a><br>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/badge/icon?subject=Java+8">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/badge/icon?subject=Java+11">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/badge/icon?subject=Portable">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/badge/icon?subject=Portable+Streaming">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Samza/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Samza/lastCompletedBuild/badge/icon?subject=Portable">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/badge/icon?subject=Portable">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/badge/icon?subject=Structured+Streaming">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Twister2/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Twister2/lastCompletedBuild/badge/icon">
</a>
</td>
</tr>
<tr>
<td>Python</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/badge/icon?subject=V1">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/badge/icon?subject=V2">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/badge/icon?subject=ValCont">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Python_PVR_Flink_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Python_PVR_Flink_Cron/lastCompletedBuild/badge/icon?subject=Portable">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Flink/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Flink/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Samza/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Samza/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
</tr>
<tr>
<td>XLang</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Direct/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Direct/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Dataflow/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Dataflow/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Samza/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Samza/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
</tr>
</tbody>
</table>
Examples testing status on various runners
--------------------------------------------------------
<table>
<thead>
<tr>
<th>Lang</th>
<th>ULR</th>
<th>Dataflow</th>
<th>Flink</th>
<th>Samza</th>
<th>Spark</th>
<th>Twister2</th>
</tr>
</thead>
<tbody>
<tr>
<td>Go</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
<tr>
<td>Java</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Java_Examples_Dataflow_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Java_Examples_Dataflow_Cron/lastCompletedBuild/badge/icon?subject=V1">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Java_Examples_Dataflow_Java11_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Java_Examples_Dataflow_Java11_Cron/lastCompletedBuild/badge/icon?subject=V1+Java11">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_Examples_Dataflow_V2/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_Examples_Dataflow_V2/lastCompletedBuild/badge/icon?subject=V2">
</a><br>
</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
<tr>
<td>Python</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
<tr>
<td>XLang</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
</tbody>
</table>
Post-Commit SDK/Transform Integration Tests Status (on master branch)
------------------------------------------------------------------------------------------------
<table>
<thead>
<tr>
<th>Go</th>
<th>Java</th>
<th>Python</th>
</tr>
</thead>
<tbody>
<tr>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/badge/icon?subject=3.6">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/badge/icon?subject=3.7">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python38/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python38/lastCompletedBuild/badge/icon?subject=3.8">
</a>
</td>
</tr>
</tbody>
</table>
Pre-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
<table>
<thead>
<tr>
<th>---</th>
<th>Java</th>
<th>Python</th>
<th>Go</th>
<th>Website</th>
<th>Whitespace</th>
<th>Typescript</th>
</tr>
</thead>
<tbody>
<tr>
<td>Non-portable</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/badge/icon">
</a><br>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/badge/icon?subject=Tests">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/badge/icon?subject=Lint">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocker_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocker_Cron/badge/icon?subject=Docker">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocs_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocs_Cron/badge/icon?subject=Docs">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Whitespace_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Whitespace_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Typescript_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Typescript_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
</tr>
<tr>
<td>Portable</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_GoPortable_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_GoPortable_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
</tbody>
</table>
See [.test-infra/jenkins/README](https://github.com/apache/beam/blob/master/.test-infra/jenkins/README.md) for trigger phrase, status and link of all Jenkins jobs.
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] msbukal commented on pull request #15936: [BEAM-13351] FhirIO GetPatientEverything connector
Posted by GitBox <gi...@apache.org>.
msbukal commented on pull request #15936:
URL: https://github.com/apache/beam/pull/15936#issuecomment-988218994
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] y1chi commented on pull request #15936: [BEAM-13351] FhirIO GetPatientEverything connector
Posted by GitBox <gi...@apache.org>.
y1chi commented on pull request #15936:
URL: https://github.com/apache/beam/pull/15936#issuecomment-988212880
It appears that we are not running other FhirIO tests with dataflow runner v2. https://github.com/apache/beam/blob/ed9f62019b43069f0cda5feb8497b7ddc9d5a620/runners/google-cloud-dataflow-java/build.gradle#L557
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 https://github.com/apache/beam/blob/ed9f62019b43069f0cda5feb8497b7ddc9d5a620/runners/google-cloud-dataflow-java/build.gradle#L547 depends on copyGoogleCloudPlatformTestResources like https://github.com/apache/beam/blob/ed9f62019b43069f0cda5feb8497b7ddc9d5a620/runners/google-cloud-dataflow-java/build.gradle#L499 (since you need to read the resource files).
You can test it locally by running the corresponding gradle task.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] msbukal commented on pull request #15936: [BEAM-13351] FhirIO GetPatientEverything connector
Posted by GitBox <gi...@apache.org>.
msbukal commented on pull request #15936:
URL: https://github.com/apache/beam/pull/15936#issuecomment-988090928
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on a change in pull request #15936: FhirIO GetPatientEverything connector
Posted by GitBox <gi...@apache.org>.
pabloem commented on a change in pull request #15936:
URL: https://github.com/apache/beam/pull/15936#discussion_r752594621
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/FhirIO.java
##########
@@ -94,20 +96,36 @@
*
* <h3>Reading</h3>
*
- * <p>FHIR resources can be read with {@link FhirIO.Read}, which supports use cases where you have
- * a ${@link PCollection} of message IDs. This is appropriate for reading the Fhir notifications
- * from a Pub/Sub subscription with {@link PubsubIO#readStrings()} or in cases where you have a
- * manually prepared list of messages that you need to process (e.g. in a text file read with {@link
+ * <p>FHIR resources can be read with {@link FhirIO.Read}, which supports use cases where you have a
+ * ${@link PCollection} of FHIR resource names in the format of projects/{p}/locations/{l}/datasets/{d}/fhirStores/{f}/fhir/{resourceType}/{id}.
+ * This is appropriate for reading the Fhir notifications from
+ * a Pub/Sub subscription with {@link PubsubIO#readStrings()} or in cases where you have a manually
+ * prepared list of resources that you need to process (e.g. in a text file read with {@link
* org.apache.beam.sdk.io.TextIO}*) .
*
- * <p>Fetch Resource contents from Fhir Store based on the {@link PCollection} of message ID
- * strings {@link FhirIO.Read.Result} where one can call {@link Read.Result#getResources()} to
- * 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
Review comment:
oops there seem to be some issues with the merged result of the file
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/FhirIOPatientEverything.java
##########
@@ -0,0 +1,317 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.healthcare;
+
+import static org.apache.beam.sdk.io.gcp.healthcare.FhirIO.BASE_METRIC_PREFIX;
+
+import com.google.gson.JsonArray;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.io.Serializable;
+import java.time.Instant;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Objects;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.coders.CustomCoder;
+import org.apache.beam.sdk.coders.DefaultCoder;
+import org.apache.beam.sdk.coders.MapCoder;
+import org.apache.beam.sdk.coders.NullableCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.io.gcp.healthcare.FhirIOPatientEverything.PatientEverythingParameter;
+import org.apache.beam.sdk.io.gcp.healthcare.HttpHealthcareApiClient.FhirResourcePagesIterator;
+import org.apache.beam.sdk.metrics.Counter;
+import org.apache.beam.sdk.metrics.Distribution;
+import org.apache.beam.sdk.metrics.Metrics;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionTuple;
+import org.apache.beam.sdk.values.PInput;
+import org.apache.beam.sdk.values.POutput;
+import org.apache.beam.sdk.values.PValue;
+import org.apache.beam.sdk.values.TupleTag;
+import org.apache.beam.sdk.values.TupleTagList;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Throwables;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableMap;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** The type FhirIOPatientEverything for querying a FHIR Patient resource's compartment. * */
+public class FhirIOPatientEverything
+ extends PTransform<PCollection<PatientEverythingParameter>, FhirIOPatientEverything.Result> {
+
+ /** The tag for the main output of FHIR Resources from a GetPatientEverything request. */
+ public static final TupleTag<JsonArray> OUT = new TupleTag<JsonArray>() {};
+ /** The tag for the deadletter output of FHIR Resources from a GetPatientEverything request. */
+ public static final TupleTag<HealthcareIOError<String>> DEAD_LETTER =
+ new TupleTag<HealthcareIOError<String>>() {};
+
+ /**
+ * PatientEverythingParameter defines required attributes for a FHIR GetPatientEverything request
+ * in {@link FhirIOPatientEverything}. *
+ */
+ @DefaultCoder(PatientEverythingParameterCoder.class)
Review comment:
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.
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/FhirIOPatientEverything.java
##########
@@ -0,0 +1,317 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.healthcare;
+
+import static org.apache.beam.sdk.io.gcp.healthcare.FhirIO.BASE_METRIC_PREFIX;
+
+import com.google.gson.JsonArray;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.io.Serializable;
+import java.time.Instant;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Objects;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.coders.CustomCoder;
+import org.apache.beam.sdk.coders.DefaultCoder;
+import org.apache.beam.sdk.coders.MapCoder;
+import org.apache.beam.sdk.coders.NullableCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.io.gcp.healthcare.FhirIOPatientEverything.PatientEverythingParameter;
+import org.apache.beam.sdk.io.gcp.healthcare.HttpHealthcareApiClient.FhirResourcePagesIterator;
+import org.apache.beam.sdk.metrics.Counter;
+import org.apache.beam.sdk.metrics.Distribution;
+import org.apache.beam.sdk.metrics.Metrics;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionTuple;
+import org.apache.beam.sdk.values.PInput;
+import org.apache.beam.sdk.values.POutput;
+import org.apache.beam.sdk.values.PValue;
+import org.apache.beam.sdk.values.TupleTag;
+import org.apache.beam.sdk.values.TupleTagList;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Throwables;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableMap;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** The type FhirIOPatientEverything for querying a FHIR Patient resource's compartment. * */
+public class FhirIOPatientEverything
+ extends PTransform<PCollection<PatientEverythingParameter>, FhirIOPatientEverything.Result> {
+
+ /** The tag for the main output of FHIR Resources from a GetPatientEverything request. */
+ public static final TupleTag<JsonArray> OUT = new TupleTag<JsonArray>() {};
+ /** The tag for the deadletter output of FHIR Resources from a GetPatientEverything request. */
+ public static final TupleTag<HealthcareIOError<String>> DEAD_LETTER =
+ new TupleTag<HealthcareIOError<String>>() {};
+
+ /**
+ * PatientEverythingParameter defines required attributes for a FHIR GetPatientEverything request
+ * in {@link FhirIOPatientEverything}. *
+ */
+ @DefaultCoder(PatientEverythingParameterCoder.class)
+ public static class PatientEverythingParameter implements Serializable {
+
+ /**
+ * FHIR Patient resource name in the format of
+ * projects/{p}/locations/{l}/datasets/{d}/fhirStores/{f}/fhir/{resourceType}/{id}.
+ */
+ private final String resourceName;
+ /** Optional filters for the request, eg. start, end, _type, _since, _count */
+ private final @Nullable Map<String, String> filters;
+
+ PatientEverythingParameter(String resourceName, @Nullable Map<String, String> filters) {
+ this.resourceName = resourceName;
+ this.filters = filters;
+ }
+
+ /**
+ * Creates a PatientEverythingParameter.
+ *
+ * @param resourceName The FHIR Patient resource name in the format of
+ * projects/{p}/locations/{l}/datasets/{d}/fhirStores/{f}/fhir/{resourceType}/{id}.
+ * @return the PatientEverythingParameter
+ */
+ public static PatientEverythingParameter of(String resourceName) {
+ return new PatientEverythingParameter(resourceName, null);
+ }
+
+ /**
+ * Creates a PatientEverythingParameter.
+ *
+ * @param resourceName The FHIR Patient resource name in the format of
+ * projects/{p}/locations/{l}/datasets/{d}/fhirStores/{f}/fhir/{resourceType}/{id}.
+ * @param filters Optional filters for the request.
+ * @return the PatientEverythingParameter
+ */
+ public static PatientEverythingParameter of(
+ String resourceName, @Nullable Map<String, String> filters) {
+ return new PatientEverythingParameter(resourceName, filters);
+ }
+
+ public String getResourceName() {
+ return resourceName;
+ }
+
+ public @Nullable Map<String, String> getFilters() {
+ return filters;
+ }
+
+ @Override
+ public boolean equals(@Nullable Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ PatientEverythingParameter that = (PatientEverythingParameter) o;
+ return Objects.equals(resourceName, that.getResourceName())
+ && Objects.equals(filters, that.getFilters());
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(resourceName, filters);
+ }
+
+ @Override
+ public String toString() {
+ return String.format(
+ "FhirIOPatientEverything.Parameter{resourcePath='%s', filters='%s'}'",
+ resourceName, filters);
+ }
+ }
+
+ /** PatientEverythingParameterCoder is a coder for {@link PatientEverythingParameter}. */
+ public static class PatientEverythingParameterCoder
+ extends CustomCoder<PatientEverythingParameter> {
+
+ private static final NullableCoder<String> STRING_CODER =
+ NullableCoder.of(StringUtf8Coder.of());
+ private static final NullableCoder<Map<String, String>> MAP_CODER =
+ NullableCoder.of(MapCoder.of(StringUtf8Coder.of(), StringUtf8Coder.of()));
+
+ public static PatientEverythingParameterCoder of() {
+ return new PatientEverythingParameterCoder();
+ }
+
+ @Override
+ public void encode(PatientEverythingParameter value, OutputStream outStream)
+ throws IOException {
+ STRING_CODER.encode(value.getResourceName(), outStream);
+ MAP_CODER.encode(value.getFilters(), outStream);
+ }
+
+ @Override
+ public PatientEverythingParameter decode(InputStream inStream) throws IOException {
+ String resourceName = STRING_CODER.decode(inStream);
+ Map<String, String> queries = MAP_CODER.decode(inStream);
+ return PatientEverythingParameter.of(resourceName, queries);
+ }
+ }
+
+ /** The Result for a {@link FhirIOPatientEverything} request. */
+ public static class Result implements POutput, PInput {
+
+ private final PCollection<JsonArray> patientCompartments;
+ private final PCollection<HealthcareIOError<String>> failedReads;
+
+ PCollectionTuple pct;
+
+ /**
+ * Create FhirIOPatientEverything.Result form PCollectionTuple with OUT and DEAD_LETTER tags.
+ *
+ * @param pct the pct
+ * @return the patient everything result
+ * @throws IllegalArgumentException the illegal argument exception
+ */
+ static Result of(PCollectionTuple pct) throws IllegalArgumentException {
+ if (pct.has(OUT) && pct.has(DEAD_LETTER)) {
+ return new Result(pct);
+ } else {
+ throw new IllegalArgumentException(
+ "The PCollection tuple must have the FhirIOPatientEverything.OUT "
+ + "and FhirIOPatientEverything.DEAD_LETTER tuple tags");
+ }
+ }
+
+ private Result(PCollectionTuple pct) {
+ this.pct = pct;
+ this.patientCompartments = pct.get(OUT).setCoder(JsonArrayCoder.of());
+ this.failedReads =
+ pct.get(DEAD_LETTER).setCoder(HealthcareIOErrorCoder.of(StringUtf8Coder.of()));
+ }
+
+ /**
+ * Gets failed reads.
+ *
+ * @return the failed reads
+ */
+ public PCollection<HealthcareIOError<String>> getFailedReads() {
+ return failedReads;
+ }
+
+ /**
+ * Gets the patient compartment responses for GetPatientEverything requests.
+ *
+ * @return the read patient compartments
+ */
+ public PCollection<JsonArray> getPatientCompartments() {
+ return patientCompartments;
+ }
+
+ @Override
+ public Pipeline getPipeline() {
+ return this.pct.getPipeline();
+ }
+
+ @Override
+ public Map<TupleTag<?>, PValue> expand() {
+ return ImmutableMap.of(OUT, patientCompartments, DEAD_LETTER, failedReads);
+ }
+
+ @Override
+ public void finishSpecifyingOutput(
+ String transformName, PInput input, PTransform<?, ?> transform) {}
+ }
+
+ @Override
+ public Result expand(PCollection<PatientEverythingParameter> input) {
+ PCollectionTuple results =
+ input.apply(
+ "GetPatientEverything",
+ ParDo.of(new GetPatientEverythingFn())
+ .withOutputTags(OUT, TupleTagList.of(DEAD_LETTER)));
+ return new Result(results);
+ }
+
+ /** GetPatientEverythingFn executes a GetPatientEverything request. */
+ static class GetPatientEverythingFn extends DoFn<PatientEverythingParameter, JsonArray> {
+
+ private static final Counter GET_PATIENT_EVERYTHING_ERROR_COUNT =
+ Metrics.counter(
+ GetPatientEverythingFn.class,
+ BASE_METRIC_PREFIX + "get_patient_everything_error_count");
+ private static final Counter GET_PATIENT_EVERYTHING_SUCCESS_COUNT =
+ Metrics.counter(
+ GetPatientEverythingFn.class,
+ BASE_METRIC_PREFIX + "get_patient_everything_success_count");
+ private static final Distribution GET_PATIENT_EVERYTHING_LATENCY_MS =
+ Metrics.distribution(
+ GetPatientEverythingFn.class, BASE_METRIC_PREFIX + "get_patient_everything_latency_ms");
+ private static final Logger LOG = LoggerFactory.getLogger(GetPatientEverythingFn.class);
+
+ @SuppressWarnings("initialization.fields.uninitialized")
+ private HealthcareApiClient client;
+
+ /**
+ * Instantiate healthcare client.
+ *
+ * @throws IOException the io exception
+ */
+ @Setup
+ public void instantiateHealthcareClient() throws IOException {
+ this.client = new HttpHealthcareApiClient();
+ }
+
+ @ProcessElement
+ public void processElement(ProcessContext context) {
+ PatientEverythingParameter patientEverythingParameter = context.element();
+ try {
+ context.output(
Review comment:
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
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/FhirIOPatientEverything.java
##########
@@ -0,0 +1,317 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.healthcare;
+
+import static org.apache.beam.sdk.io.gcp.healthcare.FhirIO.BASE_METRIC_PREFIX;
+
+import com.google.gson.JsonArray;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.io.Serializable;
+import java.time.Instant;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Objects;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.coders.CustomCoder;
+import org.apache.beam.sdk.coders.DefaultCoder;
+import org.apache.beam.sdk.coders.MapCoder;
+import org.apache.beam.sdk.coders.NullableCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.io.gcp.healthcare.FhirIOPatientEverything.PatientEverythingParameter;
+import org.apache.beam.sdk.io.gcp.healthcare.HttpHealthcareApiClient.FhirResourcePagesIterator;
+import org.apache.beam.sdk.metrics.Counter;
+import org.apache.beam.sdk.metrics.Distribution;
+import org.apache.beam.sdk.metrics.Metrics;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionTuple;
+import org.apache.beam.sdk.values.PInput;
+import org.apache.beam.sdk.values.POutput;
+import org.apache.beam.sdk.values.PValue;
+import org.apache.beam.sdk.values.TupleTag;
+import org.apache.beam.sdk.values.TupleTagList;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Throwables;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableMap;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** The type FhirIOPatientEverything for querying a FHIR Patient resource's compartment. * */
+public class FhirIOPatientEverything
+ extends PTransform<PCollection<PatientEverythingParameter>, FhirIOPatientEverything.Result> {
+
+ /** The tag for the main output of FHIR Resources from a GetPatientEverything request. */
+ public static final TupleTag<JsonArray> OUT = new TupleTag<JsonArray>() {};
+ /** The tag for the deadletter output of FHIR Resources from a GetPatientEverything request. */
+ public static final TupleTag<HealthcareIOError<String>> DEAD_LETTER =
+ new TupleTag<HealthcareIOError<String>>() {};
+
+ /**
+ * PatientEverythingParameter defines required attributes for a FHIR GetPatientEverything request
+ * in {@link FhirIOPatientEverything}. *
+ */
+ @DefaultCoder(PatientEverythingParameterCoder.class)
+ public static class PatientEverythingParameter implements Serializable {
+
+ /**
+ * FHIR Patient resource name in the format of
+ * projects/{p}/locations/{l}/datasets/{d}/fhirStores/{f}/fhir/{resourceType}/{id}.
+ */
+ private final String resourceName;
+ /** Optional filters for the request, eg. start, end, _type, _since, _count */
+ private final @Nullable Map<String, String> filters;
+
+ PatientEverythingParameter(String resourceName, @Nullable Map<String, String> filters) {
+ this.resourceName = resourceName;
+ this.filters = filters;
+ }
+
+ /**
+ * Creates a PatientEverythingParameter.
+ *
+ * @param resourceName The FHIR Patient resource name in the format of
+ * projects/{p}/locations/{l}/datasets/{d}/fhirStores/{f}/fhir/{resourceType}/{id}.
+ * @return the PatientEverythingParameter
+ */
+ public static PatientEverythingParameter of(String resourceName) {
+ return new PatientEverythingParameter(resourceName, null);
+ }
+
+ /**
+ * Creates a PatientEverythingParameter.
+ *
+ * @param resourceName The FHIR Patient resource name in the format of
+ * projects/{p}/locations/{l}/datasets/{d}/fhirStores/{f}/fhir/{resourceType}/{id}.
+ * @param filters Optional filters for the request.
+ * @return the PatientEverythingParameter
+ */
+ public static PatientEverythingParameter of(
+ String resourceName, @Nullable Map<String, String> filters) {
+ return new PatientEverythingParameter(resourceName, filters);
+ }
+
+ public String getResourceName() {
+ return resourceName;
+ }
+
+ public @Nullable Map<String, String> getFilters() {
+ return filters;
+ }
+
+ @Override
+ public boolean equals(@Nullable Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ PatientEverythingParameter that = (PatientEverythingParameter) o;
+ return Objects.equals(resourceName, that.getResourceName())
+ && Objects.equals(filters, that.getFilters());
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(resourceName, filters);
+ }
+
+ @Override
+ public String toString() {
+ return String.format(
+ "FhirIOPatientEverything.Parameter{resourcePath='%s', filters='%s'}'",
+ resourceName, filters);
+ }
+ }
+
+ /** PatientEverythingParameterCoder is a coder for {@link PatientEverythingParameter}. */
+ public static class PatientEverythingParameterCoder
+ extends CustomCoder<PatientEverythingParameter> {
+
+ private static final NullableCoder<String> STRING_CODER =
+ NullableCoder.of(StringUtf8Coder.of());
+ private static final NullableCoder<Map<String, String>> MAP_CODER =
+ NullableCoder.of(MapCoder.of(StringUtf8Coder.of(), StringUtf8Coder.of()));
+
+ public static PatientEverythingParameterCoder of() {
+ return new PatientEverythingParameterCoder();
+ }
+
+ @Override
+ public void encode(PatientEverythingParameter value, OutputStream outStream)
+ throws IOException {
+ STRING_CODER.encode(value.getResourceName(), outStream);
+ MAP_CODER.encode(value.getFilters(), outStream);
+ }
+
+ @Override
+ public PatientEverythingParameter decode(InputStream inStream) throws IOException {
+ String resourceName = STRING_CODER.decode(inStream);
+ Map<String, String> queries = MAP_CODER.decode(inStream);
+ return PatientEverythingParameter.of(resourceName, queries);
+ }
+ }
+
+ /** The Result for a {@link FhirIOPatientEverything} request. */
+ public static class Result implements POutput, PInput {
+
+ private final PCollection<JsonArray> patientCompartments;
+ private final PCollection<HealthcareIOError<String>> failedReads;
+
+ PCollectionTuple pct;
+
+ /**
+ * Create FhirIOPatientEverything.Result form PCollectionTuple with OUT and DEAD_LETTER tags.
+ *
+ * @param pct the pct
+ * @return the patient everything result
+ * @throws IllegalArgumentException the illegal argument exception
+ */
+ static Result of(PCollectionTuple pct) throws IllegalArgumentException {
+ if (pct.has(OUT) && pct.has(DEAD_LETTER)) {
+ return new Result(pct);
+ } else {
+ throw new IllegalArgumentException(
+ "The PCollection tuple must have the FhirIOPatientEverything.OUT "
+ + "and FhirIOPatientEverything.DEAD_LETTER tuple tags");
+ }
+ }
+
+ private Result(PCollectionTuple pct) {
+ this.pct = pct;
+ this.patientCompartments = pct.get(OUT).setCoder(JsonArrayCoder.of());
+ this.failedReads =
+ pct.get(DEAD_LETTER).setCoder(HealthcareIOErrorCoder.of(StringUtf8Coder.of()));
+ }
+
+ /**
+ * Gets failed reads.
+ *
+ * @return the failed reads
+ */
+ public PCollection<HealthcareIOError<String>> getFailedReads() {
+ return failedReads;
+ }
+
+ /**
+ * Gets the patient compartment responses for GetPatientEverything requests.
+ *
+ * @return the read patient compartments
+ */
+ public PCollection<JsonArray> getPatientCompartments() {
+ return patientCompartments;
+ }
+
+ @Override
+ public Pipeline getPipeline() {
+ return this.pct.getPipeline();
+ }
+
+ @Override
+ public Map<TupleTag<?>, PValue> expand() {
+ return ImmutableMap.of(OUT, patientCompartments, DEAD_LETTER, failedReads);
+ }
+
+ @Override
+ public void finishSpecifyingOutput(
+ String transformName, PInput input, PTransform<?, ?> transform) {}
+ }
+
+ @Override
+ public Result expand(PCollection<PatientEverythingParameter> input) {
+ PCollectionTuple results =
+ input.apply(
+ "GetPatientEverything",
+ ParDo.of(new GetPatientEverythingFn())
+ .withOutputTags(OUT, TupleTagList.of(DEAD_LETTER)));
+ return new Result(results);
+ }
+
+ /** GetPatientEverythingFn executes a GetPatientEverything request. */
+ static class GetPatientEverythingFn extends DoFn<PatientEverythingParameter, JsonArray> {
+
+ private static final Counter GET_PATIENT_EVERYTHING_ERROR_COUNT =
+ Metrics.counter(
+ GetPatientEverythingFn.class,
+ BASE_METRIC_PREFIX + "get_patient_everything_error_count");
+ private static final Counter GET_PATIENT_EVERYTHING_SUCCESS_COUNT =
+ Metrics.counter(
+ GetPatientEverythingFn.class,
+ BASE_METRIC_PREFIX + "get_patient_everything_success_count");
+ private static final Distribution GET_PATIENT_EVERYTHING_LATENCY_MS =
+ Metrics.distribution(
+ GetPatientEverythingFn.class, BASE_METRIC_PREFIX + "get_patient_everything_latency_ms");
+ private static final Logger LOG = LoggerFactory.getLogger(GetPatientEverythingFn.class);
+
+ @SuppressWarnings("initialization.fields.uninitialized")
+ private HealthcareApiClient client;
+
+ /**
+ * Instantiate healthcare client.
+ *
+ * @throws IOException the io exception
+ */
+ @Setup
+ public void instantiateHealthcareClient() throws IOException {
+ this.client = new HttpHealthcareApiClient();
+ }
+
+ @ProcessElement
+ public void processElement(ProcessContext context) {
+ PatientEverythingParameter patientEverythingParameter = context.element();
+ try {
+ context.output(
+ getPatientEverything(
+ patientEverythingParameter.getResourceName(),
+ patientEverythingParameter.getFilters()));
+ } catch (IllegalArgumentException | NoSuchElementException e) {
Review comment:
IIUC - if there are any other exceptions, they will be thrown out right? And retried by the runner however it chooses to - correct?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on a change in pull request #15936: FhirIO GetPatientEverything connector
Posted by GitBox <gi...@apache.org>.
pabloem commented on a change in pull request #15936:
URL: https://github.com/apache/beam/pull/15936#discussion_r754771950
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/FhirIO.java
##########
@@ -1339,13 +1353,29 @@ public void importBatch(FinishBundleContext context) throws IOException {
this.fhirStore = fhirStore;
}
- /**
- * Instantiates a new Execute bundles.
- *
- * @param fhirStore the fhir store
- */
- ExecuteBundles(String fhirStore) {
- this.fhirStore = StaticValueProvider.of(fhirStore);
+ @Override
+ ValueProvider<String> getFhirStore() {
+ return null;
Review comment:
shouldn't you return the fhirStore here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] msbukal commented on pull request #15936: FhirIO GetPatientEverything connector
Posted by GitBox <gi...@apache.org>.
msbukal commented on pull request #15936:
URL: https://github.com/apache/beam/pull/15936#issuecomment-966598277
Run Java PostCommit
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] msbukal commented on a change in pull request #15936: FhirIO GetPatientEverything connector
Posted by GitBox <gi...@apache.org>.
msbukal commented on a change in pull request #15936:
URL: https://github.com/apache/beam/pull/15936#discussion_r753547920
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/FhirIOPatientEverything.java
##########
@@ -0,0 +1,317 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.healthcare;
+
+import static org.apache.beam.sdk.io.gcp.healthcare.FhirIO.BASE_METRIC_PREFIX;
+
+import com.google.gson.JsonArray;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.io.Serializable;
+import java.time.Instant;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Objects;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.coders.CustomCoder;
+import org.apache.beam.sdk.coders.DefaultCoder;
+import org.apache.beam.sdk.coders.MapCoder;
+import org.apache.beam.sdk.coders.NullableCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.io.gcp.healthcare.FhirIOPatientEverything.PatientEverythingParameter;
+import org.apache.beam.sdk.io.gcp.healthcare.HttpHealthcareApiClient.FhirResourcePagesIterator;
+import org.apache.beam.sdk.metrics.Counter;
+import org.apache.beam.sdk.metrics.Distribution;
+import org.apache.beam.sdk.metrics.Metrics;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionTuple;
+import org.apache.beam.sdk.values.PInput;
+import org.apache.beam.sdk.values.POutput;
+import org.apache.beam.sdk.values.PValue;
+import org.apache.beam.sdk.values.TupleTag;
+import org.apache.beam.sdk.values.TupleTagList;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Throwables;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableMap;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** The type FhirIOPatientEverything for querying a FHIR Patient resource's compartment. * */
+public class FhirIOPatientEverything
+ extends PTransform<PCollection<PatientEverythingParameter>, FhirIOPatientEverything.Result> {
+
+ /** The tag for the main output of FHIR Resources from a GetPatientEverything request. */
+ public static final TupleTag<JsonArray> OUT = new TupleTag<JsonArray>() {};
+ /** The tag for the deadletter output of FHIR Resources from a GetPatientEverything request. */
+ public static final TupleTag<HealthcareIOError<String>> DEAD_LETTER =
+ new TupleTag<HealthcareIOError<String>>() {};
+
+ /**
+ * PatientEverythingParameter defines required attributes for a FHIR GetPatientEverything request
+ * in {@link FhirIOPatientEverything}. *
+ */
+ @DefaultCoder(PatientEverythingParameterCoder.class)
+ public static class PatientEverythingParameter implements Serializable {
+
+ /**
+ * FHIR Patient resource name in the format of
+ * projects/{p}/locations/{l}/datasets/{d}/fhirStores/{f}/fhir/{resourceType}/{id}.
+ */
+ private final String resourceName;
+ /** Optional filters for the request, eg. start, end, _type, _since, _count */
+ private final @Nullable Map<String, String> filters;
+
+ PatientEverythingParameter(String resourceName, @Nullable Map<String, String> filters) {
+ this.resourceName = resourceName;
+ this.filters = filters;
+ }
+
+ /**
+ * Creates a PatientEverythingParameter.
+ *
+ * @param resourceName The FHIR Patient resource name in the format of
+ * projects/{p}/locations/{l}/datasets/{d}/fhirStores/{f}/fhir/{resourceType}/{id}.
+ * @return the PatientEverythingParameter
+ */
+ public static PatientEverythingParameter of(String resourceName) {
+ return new PatientEverythingParameter(resourceName, null);
+ }
+
+ /**
+ * Creates a PatientEverythingParameter.
+ *
+ * @param resourceName The FHIR Patient resource name in the format of
+ * projects/{p}/locations/{l}/datasets/{d}/fhirStores/{f}/fhir/{resourceType}/{id}.
+ * @param filters Optional filters for the request.
+ * @return the PatientEverythingParameter
+ */
+ public static PatientEverythingParameter of(
+ String resourceName, @Nullable Map<String, String> filters) {
+ return new PatientEverythingParameter(resourceName, filters);
+ }
+
+ public String getResourceName() {
+ return resourceName;
+ }
+
+ public @Nullable Map<String, String> getFilters() {
+ return filters;
+ }
+
+ @Override
+ public boolean equals(@Nullable Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ PatientEverythingParameter that = (PatientEverythingParameter) o;
+ return Objects.equals(resourceName, that.getResourceName())
+ && Objects.equals(filters, that.getFilters());
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(resourceName, filters);
+ }
+
+ @Override
+ public String toString() {
+ return String.format(
+ "FhirIOPatientEverything.Parameter{resourcePath='%s', filters='%s'}'",
+ resourceName, filters);
+ }
+ }
+
+ /** PatientEverythingParameterCoder is a coder for {@link PatientEverythingParameter}. */
+ public static class PatientEverythingParameterCoder
+ extends CustomCoder<PatientEverythingParameter> {
+
+ private static final NullableCoder<String> STRING_CODER =
+ NullableCoder.of(StringUtf8Coder.of());
+ private static final NullableCoder<Map<String, String>> MAP_CODER =
+ NullableCoder.of(MapCoder.of(StringUtf8Coder.of(), StringUtf8Coder.of()));
+
+ public static PatientEverythingParameterCoder of() {
+ return new PatientEverythingParameterCoder();
+ }
+
+ @Override
+ public void encode(PatientEverythingParameter value, OutputStream outStream)
+ throws IOException {
+ STRING_CODER.encode(value.getResourceName(), outStream);
+ MAP_CODER.encode(value.getFilters(), outStream);
+ }
+
+ @Override
+ public PatientEverythingParameter decode(InputStream inStream) throws IOException {
+ String resourceName = STRING_CODER.decode(inStream);
+ Map<String, String> queries = MAP_CODER.decode(inStream);
+ return PatientEverythingParameter.of(resourceName, queries);
+ }
+ }
+
+ /** The Result for a {@link FhirIOPatientEverything} request. */
+ public static class Result implements POutput, PInput {
+
+ private final PCollection<JsonArray> patientCompartments;
+ private final PCollection<HealthcareIOError<String>> failedReads;
+
+ PCollectionTuple pct;
+
+ /**
+ * Create FhirIOPatientEverything.Result form PCollectionTuple with OUT and DEAD_LETTER tags.
+ *
+ * @param pct the pct
+ * @return the patient everything result
+ * @throws IllegalArgumentException the illegal argument exception
+ */
+ static Result of(PCollectionTuple pct) throws IllegalArgumentException {
+ if (pct.has(OUT) && pct.has(DEAD_LETTER)) {
+ return new Result(pct);
+ } else {
+ throw new IllegalArgumentException(
+ "The PCollection tuple must have the FhirIOPatientEverything.OUT "
+ + "and FhirIOPatientEverything.DEAD_LETTER tuple tags");
+ }
+ }
+
+ private Result(PCollectionTuple pct) {
+ this.pct = pct;
+ this.patientCompartments = pct.get(OUT).setCoder(JsonArrayCoder.of());
+ this.failedReads =
+ pct.get(DEAD_LETTER).setCoder(HealthcareIOErrorCoder.of(StringUtf8Coder.of()));
+ }
+
+ /**
+ * Gets failed reads.
+ *
+ * @return the failed reads
+ */
+ public PCollection<HealthcareIOError<String>> getFailedReads() {
+ return failedReads;
+ }
+
+ /**
+ * Gets the patient compartment responses for GetPatientEverything requests.
+ *
+ * @return the read patient compartments
+ */
+ public PCollection<JsonArray> getPatientCompartments() {
+ return patientCompartments;
+ }
+
+ @Override
+ public Pipeline getPipeline() {
+ return this.pct.getPipeline();
+ }
+
+ @Override
+ public Map<TupleTag<?>, PValue> expand() {
+ return ImmutableMap.of(OUT, patientCompartments, DEAD_LETTER, failedReads);
+ }
+
+ @Override
+ public void finishSpecifyingOutput(
+ String transformName, PInput input, PTransform<?, ?> transform) {}
+ }
+
+ @Override
+ public Result expand(PCollection<PatientEverythingParameter> input) {
+ PCollectionTuple results =
+ input.apply(
+ "GetPatientEverything",
+ ParDo.of(new GetPatientEverythingFn())
+ .withOutputTags(OUT, TupleTagList.of(DEAD_LETTER)));
+ return new Result(results);
+ }
+
+ /** GetPatientEverythingFn executes a GetPatientEverything request. */
+ static class GetPatientEverythingFn extends DoFn<PatientEverythingParameter, JsonArray> {
+
+ private static final Counter GET_PATIENT_EVERYTHING_ERROR_COUNT =
+ Metrics.counter(
+ GetPatientEverythingFn.class,
+ BASE_METRIC_PREFIX + "get_patient_everything_error_count");
+ private static final Counter GET_PATIENT_EVERYTHING_SUCCESS_COUNT =
+ Metrics.counter(
+ GetPatientEverythingFn.class,
+ BASE_METRIC_PREFIX + "get_patient_everything_success_count");
+ private static final Distribution GET_PATIENT_EVERYTHING_LATENCY_MS =
+ Metrics.distribution(
+ GetPatientEverythingFn.class, BASE_METRIC_PREFIX + "get_patient_everything_latency_ms");
+ private static final Logger LOG = LoggerFactory.getLogger(GetPatientEverythingFn.class);
+
+ @SuppressWarnings("initialization.fields.uninitialized")
+ private HealthcareApiClient client;
+
+ /**
+ * Instantiate healthcare client.
+ *
+ * @throws IOException the io exception
+ */
+ @Setup
+ public void instantiateHealthcareClient() throws IOException {
+ this.client = new HttpHealthcareApiClient();
+ }
+
+ @ProcessElement
+ public void processElement(ProcessContext context) {
+ PatientEverythingParameter patientEverythingParameter = context.element();
+ try {
+ context.output(
+ getPatientEverything(
+ patientEverythingParameter.getResourceName(),
+ patientEverythingParameter.getFilters()));
+ } catch (IllegalArgumentException | NoSuchElementException e) {
Review comment:
Yes, that's what I believe.
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/FhirIO.java
##########
@@ -94,20 +96,36 @@
*
* <h3>Reading</h3>
*
- * <p>FHIR resources can be read with {@link FhirIO.Read}, which supports use cases where you have
- * a ${@link PCollection} of message IDs. This is appropriate for reading the Fhir notifications
- * from a Pub/Sub subscription with {@link PubsubIO#readStrings()} or in cases where you have a
- * manually prepared list of messages that you need to process (e.g. in a text file read with {@link
+ * <p>FHIR resources can be read with {@link FhirIO.Read}, which supports use cases where you have a
+ * ${@link PCollection} of FHIR resource names in the format of projects/{p}/locations/{l}/datasets/{d}/fhirStores/{f}/fhir/{resourceType}/{id}.
+ * This is appropriate for reading the Fhir notifications from
+ * a Pub/Sub subscription with {@link PubsubIO#readStrings()} or in cases where you have a manually
+ * prepared list of resources that you need to process (e.g. in a text file read with {@link
* org.apache.beam.sdk.io.TextIO}*) .
*
- * <p>Fetch Resource contents from Fhir Store based on the {@link PCollection} of message ID
- * strings {@link FhirIO.Read.Result} where one can call {@link Read.Result#getResources()} to
- * 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
Review comment:
Whoops, missed these somehow because they didn't cause runtime errors haha.
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/FhirIOPatientEverything.java
##########
@@ -0,0 +1,317 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.healthcare;
+
+import static org.apache.beam.sdk.io.gcp.healthcare.FhirIO.BASE_METRIC_PREFIX;
+
+import com.google.gson.JsonArray;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.io.Serializable;
+import java.time.Instant;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Objects;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.coders.CustomCoder;
+import org.apache.beam.sdk.coders.DefaultCoder;
+import org.apache.beam.sdk.coders.MapCoder;
+import org.apache.beam.sdk.coders.NullableCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.io.gcp.healthcare.FhirIOPatientEverything.PatientEverythingParameter;
+import org.apache.beam.sdk.io.gcp.healthcare.HttpHealthcareApiClient.FhirResourcePagesIterator;
+import org.apache.beam.sdk.metrics.Counter;
+import org.apache.beam.sdk.metrics.Distribution;
+import org.apache.beam.sdk.metrics.Metrics;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionTuple;
+import org.apache.beam.sdk.values.PInput;
+import org.apache.beam.sdk.values.POutput;
+import org.apache.beam.sdk.values.PValue;
+import org.apache.beam.sdk.values.TupleTag;
+import org.apache.beam.sdk.values.TupleTagList;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Throwables;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableMap;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** The type FhirIOPatientEverything for querying a FHIR Patient resource's compartment. * */
+public class FhirIOPatientEverything
+ extends PTransform<PCollection<PatientEverythingParameter>, FhirIOPatientEverything.Result> {
+
+ /** The tag for the main output of FHIR Resources from a GetPatientEverything request. */
+ public static final TupleTag<JsonArray> OUT = new TupleTag<JsonArray>() {};
+ /** The tag for the deadletter output of FHIR Resources from a GetPatientEverything request. */
+ public static final TupleTag<HealthcareIOError<String>> DEAD_LETTER =
+ new TupleTag<HealthcareIOError<String>>() {};
+
+ /**
+ * PatientEverythingParameter defines required attributes for a FHIR GetPatientEverything request
+ * in {@link FhirIOPatientEverything}. *
+ */
+ @DefaultCoder(PatientEverythingParameterCoder.class)
Review comment:
Done, I made them AutoValue!
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/FhirIOPatientEverything.java
##########
@@ -0,0 +1,317 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.healthcare;
+
+import static org.apache.beam.sdk.io.gcp.healthcare.FhirIO.BASE_METRIC_PREFIX;
+
+import com.google.gson.JsonArray;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+import java.io.Serializable;
+import java.time.Instant;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.Objects;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.coders.CustomCoder;
+import org.apache.beam.sdk.coders.DefaultCoder;
+import org.apache.beam.sdk.coders.MapCoder;
+import org.apache.beam.sdk.coders.NullableCoder;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.io.gcp.healthcare.FhirIOPatientEverything.PatientEverythingParameter;
+import org.apache.beam.sdk.io.gcp.healthcare.HttpHealthcareApiClient.FhirResourcePagesIterator;
+import org.apache.beam.sdk.metrics.Counter;
+import org.apache.beam.sdk.metrics.Distribution;
+import org.apache.beam.sdk.metrics.Metrics;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionTuple;
+import org.apache.beam.sdk.values.PInput;
+import org.apache.beam.sdk.values.POutput;
+import org.apache.beam.sdk.values.PValue;
+import org.apache.beam.sdk.values.TupleTag;
+import org.apache.beam.sdk.values.TupleTagList;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Throwables;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableMap;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** The type FhirIOPatientEverything for querying a FHIR Patient resource's compartment. * */
+public class FhirIOPatientEverything
+ extends PTransform<PCollection<PatientEverythingParameter>, FhirIOPatientEverything.Result> {
+
+ /** The tag for the main output of FHIR Resources from a GetPatientEverything request. */
+ public static final TupleTag<JsonArray> OUT = new TupleTag<JsonArray>() {};
+ /** The tag for the deadletter output of FHIR Resources from a GetPatientEverything request. */
+ public static final TupleTag<HealthcareIOError<String>> DEAD_LETTER =
+ new TupleTag<HealthcareIOError<String>>() {};
+
+ /**
+ * PatientEverythingParameter defines required attributes for a FHIR GetPatientEverything request
+ * in {@link FhirIOPatientEverything}. *
+ */
+ @DefaultCoder(PatientEverythingParameterCoder.class)
+ public static class PatientEverythingParameter implements Serializable {
+
+ /**
+ * FHIR Patient resource name in the format of
+ * projects/{p}/locations/{l}/datasets/{d}/fhirStores/{f}/fhir/{resourceType}/{id}.
+ */
+ private final String resourceName;
+ /** Optional filters for the request, eg. start, end, _type, _since, _count */
+ private final @Nullable Map<String, String> filters;
+
+ PatientEverythingParameter(String resourceName, @Nullable Map<String, String> filters) {
+ this.resourceName = resourceName;
+ this.filters = filters;
+ }
+
+ /**
+ * Creates a PatientEverythingParameter.
+ *
+ * @param resourceName The FHIR Patient resource name in the format of
+ * projects/{p}/locations/{l}/datasets/{d}/fhirStores/{f}/fhir/{resourceType}/{id}.
+ * @return the PatientEverythingParameter
+ */
+ public static PatientEverythingParameter of(String resourceName) {
+ return new PatientEverythingParameter(resourceName, null);
+ }
+
+ /**
+ * Creates a PatientEverythingParameter.
+ *
+ * @param resourceName The FHIR Patient resource name in the format of
+ * projects/{p}/locations/{l}/datasets/{d}/fhirStores/{f}/fhir/{resourceType}/{id}.
+ * @param filters Optional filters for the request.
+ * @return the PatientEverythingParameter
+ */
+ public static PatientEverythingParameter of(
+ String resourceName, @Nullable Map<String, String> filters) {
+ return new PatientEverythingParameter(resourceName, filters);
+ }
+
+ public String getResourceName() {
+ return resourceName;
+ }
+
+ public @Nullable Map<String, String> getFilters() {
+ return filters;
+ }
+
+ @Override
+ public boolean equals(@Nullable Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ PatientEverythingParameter that = (PatientEverythingParameter) o;
+ return Objects.equals(resourceName, that.getResourceName())
+ && Objects.equals(filters, that.getFilters());
+ }
+
+ @Override
+ public int hashCode() {
+ return Objects.hash(resourceName, filters);
+ }
+
+ @Override
+ public String toString() {
+ return String.format(
+ "FhirIOPatientEverything.Parameter{resourcePath='%s', filters='%s'}'",
+ resourceName, filters);
+ }
+ }
+
+ /** PatientEverythingParameterCoder is a coder for {@link PatientEverythingParameter}. */
+ public static class PatientEverythingParameterCoder
+ extends CustomCoder<PatientEverythingParameter> {
+
+ private static final NullableCoder<String> STRING_CODER =
+ NullableCoder.of(StringUtf8Coder.of());
+ private static final NullableCoder<Map<String, String>> MAP_CODER =
+ NullableCoder.of(MapCoder.of(StringUtf8Coder.of(), StringUtf8Coder.of()));
+
+ public static PatientEverythingParameterCoder of() {
+ return new PatientEverythingParameterCoder();
+ }
+
+ @Override
+ public void encode(PatientEverythingParameter value, OutputStream outStream)
+ throws IOException {
+ STRING_CODER.encode(value.getResourceName(), outStream);
+ MAP_CODER.encode(value.getFilters(), outStream);
+ }
+
+ @Override
+ public PatientEverythingParameter decode(InputStream inStream) throws IOException {
+ String resourceName = STRING_CODER.decode(inStream);
+ Map<String, String> queries = MAP_CODER.decode(inStream);
+ return PatientEverythingParameter.of(resourceName, queries);
+ }
+ }
+
+ /** The Result for a {@link FhirIOPatientEverything} request. */
+ public static class Result implements POutput, PInput {
+
+ private final PCollection<JsonArray> patientCompartments;
+ private final PCollection<HealthcareIOError<String>> failedReads;
+
+ PCollectionTuple pct;
+
+ /**
+ * Create FhirIOPatientEverything.Result form PCollectionTuple with OUT and DEAD_LETTER tags.
+ *
+ * @param pct the pct
+ * @return the patient everything result
+ * @throws IllegalArgumentException the illegal argument exception
+ */
+ static Result of(PCollectionTuple pct) throws IllegalArgumentException {
+ if (pct.has(OUT) && pct.has(DEAD_LETTER)) {
+ return new Result(pct);
+ } else {
+ throw new IllegalArgumentException(
+ "The PCollection tuple must have the FhirIOPatientEverything.OUT "
+ + "and FhirIOPatientEverything.DEAD_LETTER tuple tags");
+ }
+ }
+
+ private Result(PCollectionTuple pct) {
+ this.pct = pct;
+ this.patientCompartments = pct.get(OUT).setCoder(JsonArrayCoder.of());
+ this.failedReads =
+ pct.get(DEAD_LETTER).setCoder(HealthcareIOErrorCoder.of(StringUtf8Coder.of()));
+ }
+
+ /**
+ * Gets failed reads.
+ *
+ * @return the failed reads
+ */
+ public PCollection<HealthcareIOError<String>> getFailedReads() {
+ return failedReads;
+ }
+
+ /**
+ * Gets the patient compartment responses for GetPatientEverything requests.
+ *
+ * @return the read patient compartments
+ */
+ public PCollection<JsonArray> getPatientCompartments() {
+ return patientCompartments;
+ }
+
+ @Override
+ public Pipeline getPipeline() {
+ return this.pct.getPipeline();
+ }
+
+ @Override
+ public Map<TupleTag<?>, PValue> expand() {
+ return ImmutableMap.of(OUT, patientCompartments, DEAD_LETTER, failedReads);
+ }
+
+ @Override
+ public void finishSpecifyingOutput(
+ String transformName, PInput input, PTransform<?, ?> transform) {}
+ }
+
+ @Override
+ public Result expand(PCollection<PatientEverythingParameter> input) {
+ PCollectionTuple results =
+ input.apply(
+ "GetPatientEverything",
+ ParDo.of(new GetPatientEverythingFn())
+ .withOutputTags(OUT, TupleTagList.of(DEAD_LETTER)));
+ return new Result(results);
+ }
+
+ /** GetPatientEverythingFn executes a GetPatientEverything request. */
+ static class GetPatientEverythingFn extends DoFn<PatientEverythingParameter, JsonArray> {
+
+ private static final Counter GET_PATIENT_EVERYTHING_ERROR_COUNT =
+ Metrics.counter(
+ GetPatientEverythingFn.class,
+ BASE_METRIC_PREFIX + "get_patient_everything_error_count");
+ private static final Counter GET_PATIENT_EVERYTHING_SUCCESS_COUNT =
+ Metrics.counter(
+ GetPatientEverythingFn.class,
+ BASE_METRIC_PREFIX + "get_patient_everything_success_count");
+ private static final Distribution GET_PATIENT_EVERYTHING_LATENCY_MS =
+ Metrics.distribution(
+ GetPatientEverythingFn.class, BASE_METRIC_PREFIX + "get_patient_everything_latency_ms");
+ private static final Logger LOG = LoggerFactory.getLogger(GetPatientEverythingFn.class);
+
+ @SuppressWarnings("initialization.fields.uninitialized")
+ private HealthcareApiClient client;
+
+ /**
+ * Instantiate healthcare client.
+ *
+ * @throws IOException the io exception
+ */
+ @Setup
+ public void instantiateHealthcareClient() throws IOException {
+ this.client = new HttpHealthcareApiClient();
+ }
+
+ @ProcessElement
+ public void processElement(ProcessContext context) {
+ PatientEverythingParameter patientEverythingParameter = context.element();
+ try {
+ context.output(
Review comment:
Will follow up for this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] msbukal commented on a change in pull request #15936: FhirIO GetPatientEverything connector
Posted by GitBox <gi...@apache.org>.
msbukal commented on a change in pull request #15936:
URL: https://github.com/apache/beam/pull/15936#discussion_r756339041
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/FhirIO.java
##########
@@ -1339,13 +1353,29 @@ public void importBatch(FinishBundleContext context) throws IOException {
this.fhirStore = fhirStore;
}
- /**
- * Instantiates a new Execute bundles.
- *
- * @param fhirStore the fhir store
- */
- ExecuteBundles(String fhirStore) {
- this.fhirStore = StaticValueProvider.of(fhirStore);
+ @Override
+ ValueProvider<String> getFhirStore() {
+ return null;
+ }
+
+ @Override
+ WriteMethod getWriteMethod() {
+ return WriteMethod.EXECUTE_BUNDLE;
+ }
+
+ @Override
+ Optional<ContentStructure> getContentStructure() {
+ return Optional.empty();
+ }
+
+ @Override
+ Optional<ValueProvider<String>> getImportGcsTempPath() {
+ return Optional.empty();
+ }
+
+ @Override
+ Optional<ValueProvider<String>> getImportGcsDeadLetterPath() {
+ return Optional.empty();
Review comment:
I'm not sure what you mean, these methods are only here because I made ExecuteBundles implement FhirIO.Write, which requires them.
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/FhirIOPatientEverything.java
##########
@@ -0,0 +1,237 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.healthcare;
+
+import static org.apache.beam.sdk.io.gcp.healthcare.FhirIO.BASE_METRIC_PREFIX;
+
+import com.google.auto.value.AutoValue;
+import com.google.gson.JsonArray;
+import java.io.IOException;
+import java.io.Serializable;
+import java.time.Instant;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.io.gcp.healthcare.FhirIOPatientEverything.PatientEverythingParameter;
+import org.apache.beam.sdk.io.gcp.healthcare.HttpHealthcareApiClient.FhirResourcePagesIterator;
+import org.apache.beam.sdk.metrics.Counter;
+import org.apache.beam.sdk.metrics.Distribution;
+import org.apache.beam.sdk.metrics.Metrics;
+import org.apache.beam.sdk.schemas.AutoValueSchema;
+import org.apache.beam.sdk.schemas.annotations.DefaultSchema;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionTuple;
+import org.apache.beam.sdk.values.PInput;
+import org.apache.beam.sdk.values.POutput;
+import org.apache.beam.sdk.values.PValue;
+import org.apache.beam.sdk.values.TupleTag;
+import org.apache.beam.sdk.values.TupleTagList;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Throwables;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableMap;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** The type FhirIOPatientEverything for querying a FHIR Patient resource's compartment. * */
+public class FhirIOPatientEverything
+ extends PTransform<PCollection<PatientEverythingParameter>, FhirIOPatientEverything.Result> {
+
+ /** The tag for the main output of FHIR Resources from a GetPatientEverything request. */
+ public static final TupleTag<JsonArray> OUT = new TupleTag<JsonArray>() {};
+ /** The tag for the deadletter output of FHIR Resources from a GetPatientEverything request. */
+ public static final TupleTag<HealthcareIOError<String>> DEAD_LETTER =
+ new TupleTag<HealthcareIOError<String>>() {};
+
+ /**
+ * PatientEverythingParameter defines required attributes for a FHIR GetPatientEverything request
+ * in {@link FhirIOPatientEverything}. *
+ */
+ @DefaultSchema(AutoValueSchema.class)
+ @AutoValue
+ public abstract static class PatientEverythingParameter implements Serializable {
+
+ /**
+ * FHIR Patient resource name in the format of
+ * projects/{p}/locations/{l}/datasets/{d}/fhirStores/{f}/fhir/{resourceType}/{id}.
+ */
+ abstract String getResourceName();
+ /** Optional filters for the request, eg. start, end, _type, _since, _count */
+ abstract @Nullable Map<String, String> getFilters();
+
+ static Builder builder() {
+ return new AutoValue_FhirIOPatientEverything_PatientEverythingParameter.Builder();
+ }
+
+ @AutoValue.Builder
+ abstract static class Builder {
+ abstract Builder setResourceName(String resourceName);
+
+ abstract Builder setFilters(Map<String, String> query);
+
+ abstract PatientEverythingParameter build();
+ }
+ }
+
+ /** The Result for a {@link FhirIOPatientEverything} request. */
+ public static class Result implements POutput, PInput {
+
+ private final PCollection<JsonArray> patientCompartments;
+ private final PCollection<HealthcareIOError<String>> failedReads;
+
+ PCollectionTuple pct;
+
+ /**
+ * Create FhirIOPatientEverything.Result form PCollectionTuple with OUT and DEAD_LETTER tags.
+ *
+ * @param pct the pct
+ * @return the patient everything result
+ * @throws IllegalArgumentException the illegal argument exception
+ */
+ static Result of(PCollectionTuple pct) throws IllegalArgumentException {
+ if (pct.has(OUT) && pct.has(DEAD_LETTER)) {
+ return new Result(pct);
+ } else {
+ throw new IllegalArgumentException(
+ "The PCollection tuple must have the FhirIOPatientEverything.OUT "
+ + "and FhirIOPatientEverything.DEAD_LETTER tuple tags");
+ }
+ }
+
+ private Result(PCollectionTuple pct) {
+ this.pct = pct;
+ this.patientCompartments = pct.get(OUT).setCoder(JsonArrayCoder.of());
+ this.failedReads =
+ pct.get(DEAD_LETTER).setCoder(HealthcareIOErrorCoder.of(StringUtf8Coder.of()));
+ }
+
+ /**
+ * Gets failed reads.
+ *
+ * @return the failed reads
+ */
+ public PCollection<HealthcareIOError<String>> getFailedReads() {
+ return failedReads;
+ }
+
+ /**
+ * Gets the patient compartment responses for GetPatientEverything requests.
+ *
+ * @return the read patient compartments
+ */
+ public PCollection<JsonArray> getPatientCompartments() {
+ return patientCompartments;
+ }
+
+ @Override
+ public Pipeline getPipeline() {
+ return this.pct.getPipeline();
+ }
+
+ @Override
+ public Map<TupleTag<?>, PValue> expand() {
Review comment:
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?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] msbukal commented on pull request #15936: [BEAM-13351] FhirIO GetPatientEverything connector
Posted by GitBox <gi...@apache.org>.
msbukal commented on pull request #15936:
URL: https://github.com/apache/beam/pull/15936#issuecomment-987045657
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](https://github.com/apache/beam/pull/15936/files#diff-70bd3a45eeb201c23bf129069100faea2d2ba1550f955a74e64057c28bbe021eR85) but this is the exact same setup that other integration tests use (eg. [FhirIOSearchIT](https://github.com/msbukal/beam/blob/1d21e872c33ba06260816858ccb9737fde0e8371/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/healthcare/FhirIOSearchIT.java#L92)). Probably because I run the test independently it passes, will investigate more.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on pull request #15936: FhirIO GetPatientEverything connector
Posted by GitBox <gi...@apache.org>.
pabloem commented on pull request #15936:
URL: https://github.com/apache/beam/pull/15936#issuecomment-982733376
LGTM. Sorry about the delay!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] y1chi commented on pull request #15936: [BEAM-13351] FhirIO GetPatientEverything connector
Posted by GitBox <gi...@apache.org>.
y1chi commented on pull request #15936:
URL: https://github.com/apache/beam/pull/15936#issuecomment-987033246
The test is failing and breaking post commit, https://issues.apache.org/jira/browse/BEAM-13387 @msbukal could you take a look at it?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] y1chi commented on pull request #15936: [BEAM-13351] FhirIO GetPatientEverything connector
Posted by GitBox <gi...@apache.org>.
y1chi commented on pull request #15936:
URL: https://github.com/apache/beam/pull/15936#issuecomment-987052788
> 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](https://github.com/apache/beam/pull/15936/files#diff-70bd3a45eeb201c23bf129069100faea2d2ba1550f955a74e64057c28bbe021eR85) but this is the exact same setup that other integration tests use (eg. [FhirIOSearchIT](https://github.com/msbukal/beam/blob/1d21e872c33ba06260816858ccb9737fde0e8371/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/healthcare/FhirIOSearchIT.java#L92)). 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_/
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on a change in pull request #15936: FhirIO GetPatientEverything connector
Posted by GitBox <gi...@apache.org>.
pabloem commented on a change in pull request #15936:
URL: https://github.com/apache/beam/pull/15936#discussion_r754772280
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/FhirIO.java
##########
@@ -1339,13 +1353,29 @@ public void importBatch(FinishBundleContext context) throws IOException {
this.fhirStore = fhirStore;
}
- /**
- * Instantiates a new Execute bundles.
- *
- * @param fhirStore the fhir store
- */
- ExecuteBundles(String fhirStore) {
- this.fhirStore = StaticValueProvider.of(fhirStore);
+ @Override
+ ValueProvider<String> getFhirStore() {
+ return null;
+ }
+
+ @Override
+ WriteMethod getWriteMethod() {
+ return WriteMethod.EXECUTE_BUNDLE;
+ }
+
+ @Override
+ Optional<ContentStructure> getContentStructure() {
+ return Optional.empty();
+ }
+
+ @Override
+ Optional<ValueProvider<String>> getImportGcsTempPath() {
+ return Optional.empty();
+ }
+
+ @Override
+ Optional<ValueProvider<String>> getImportGcsDeadLetterPath() {
+ return Optional.empty();
Review comment:
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<String>
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/FhirIOPatientEverything.java
##########
@@ -0,0 +1,237 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.healthcare;
+
+import static org.apache.beam.sdk.io.gcp.healthcare.FhirIO.BASE_METRIC_PREFIX;
+
+import com.google.auto.value.AutoValue;
+import com.google.gson.JsonArray;
+import java.io.IOException;
+import java.io.Serializable;
+import java.time.Instant;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.io.gcp.healthcare.FhirIOPatientEverything.PatientEverythingParameter;
+import org.apache.beam.sdk.io.gcp.healthcare.HttpHealthcareApiClient.FhirResourcePagesIterator;
+import org.apache.beam.sdk.metrics.Counter;
+import org.apache.beam.sdk.metrics.Distribution;
+import org.apache.beam.sdk.metrics.Metrics;
+import org.apache.beam.sdk.schemas.AutoValueSchema;
+import org.apache.beam.sdk.schemas.annotations.DefaultSchema;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionTuple;
+import org.apache.beam.sdk.values.PInput;
+import org.apache.beam.sdk.values.POutput;
+import org.apache.beam.sdk.values.PValue;
+import org.apache.beam.sdk.values.TupleTag;
+import org.apache.beam.sdk.values.TupleTagList;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Throwables;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableMap;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** The type FhirIOPatientEverything for querying a FHIR Patient resource's compartment. * */
+public class FhirIOPatientEverything
+ extends PTransform<PCollection<PatientEverythingParameter>, FhirIOPatientEverything.Result> {
+
+ /** The tag for the main output of FHIR Resources from a GetPatientEverything request. */
+ public static final TupleTag<JsonArray> OUT = new TupleTag<JsonArray>() {};
+ /** The tag for the deadletter output of FHIR Resources from a GetPatientEverything request. */
+ public static final TupleTag<HealthcareIOError<String>> DEAD_LETTER =
+ new TupleTag<HealthcareIOError<String>>() {};
+
+ /**
+ * PatientEverythingParameter defines required attributes for a FHIR GetPatientEverything request
+ * in {@link FhirIOPatientEverything}. *
+ */
+ @DefaultSchema(AutoValueSchema.class)
+ @AutoValue
+ public abstract static class PatientEverythingParameter implements Serializable {
+
+ /**
+ * FHIR Patient resource name in the format of
+ * projects/{p}/locations/{l}/datasets/{d}/fhirStores/{f}/fhir/{resourceType}/{id}.
+ */
+ abstract String getResourceName();
+ /** Optional filters for the request, eg. start, end, _type, _since, _count */
+ abstract @Nullable Map<String, String> getFilters();
+
+ static Builder builder() {
+ return new AutoValue_FhirIOPatientEverything_PatientEverythingParameter.Builder();
+ }
+
+ @AutoValue.Builder
+ abstract static class Builder {
+ abstract Builder setResourceName(String resourceName);
+
+ abstract Builder setFilters(Map<String, String> query);
+
+ abstract PatientEverythingParameter build();
+ }
+ }
+
+ /** The Result for a {@link FhirIOPatientEverything} request. */
+ public static class Result implements POutput, PInput {
+
+ private final PCollection<JsonArray> patientCompartments;
+ private final PCollection<HealthcareIOError<String>> failedReads;
+
+ PCollectionTuple pct;
+
+ /**
+ * Create FhirIOPatientEverything.Result form PCollectionTuple with OUT and DEAD_LETTER tags.
+ *
+ * @param pct the pct
+ * @return the patient everything result
+ * @throws IllegalArgumentException the illegal argument exception
+ */
+ static Result of(PCollectionTuple pct) throws IllegalArgumentException {
+ if (pct.has(OUT) && pct.has(DEAD_LETTER)) {
+ return new Result(pct);
+ } else {
+ throw new IllegalArgumentException(
+ "The PCollection tuple must have the FhirIOPatientEverything.OUT "
+ + "and FhirIOPatientEverything.DEAD_LETTER tuple tags");
+ }
+ }
+
+ private Result(PCollectionTuple pct) {
+ this.pct = pct;
+ this.patientCompartments = pct.get(OUT).setCoder(JsonArrayCoder.of());
+ this.failedReads =
+ pct.get(DEAD_LETTER).setCoder(HealthcareIOErrorCoder.of(StringUtf8Coder.of()));
+ }
+
+ /**
+ * Gets failed reads.
+ *
+ * @return the failed reads
+ */
+ public PCollection<HealthcareIOError<String>> getFailedReads() {
+ return failedReads;
+ }
+
+ /**
+ * Gets the patient compartment responses for GetPatientEverything requests.
+ *
+ * @return the read patient compartments
+ */
+ public PCollection<JsonArray> getPatientCompartments() {
+ return patientCompartments;
+ }
+
+ @Override
+ public Pipeline getPipeline() {
+ return this.pct.getPipeline();
+ }
+
+ @Override
+ public Map<TupleTag<?>, PValue> expand() {
Review comment:
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?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem commented on a change in pull request #15936: FhirIO GetPatientEverything connector
Posted by GitBox <gi...@apache.org>.
pabloem commented on a change in pull request #15936:
URL: https://github.com/apache/beam/pull/15936#discussion_r759381443
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/FhirIOPatientEverything.java
##########
@@ -0,0 +1,237 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.beam.sdk.io.gcp.healthcare;
+
+import static org.apache.beam.sdk.io.gcp.healthcare.FhirIO.BASE_METRIC_PREFIX;
+
+import com.google.auto.value.AutoValue;
+import com.google.gson.JsonArray;
+import java.io.IOException;
+import java.io.Serializable;
+import java.time.Instant;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.coders.StringUtf8Coder;
+import org.apache.beam.sdk.io.gcp.healthcare.FhirIOPatientEverything.PatientEverythingParameter;
+import org.apache.beam.sdk.io.gcp.healthcare.HttpHealthcareApiClient.FhirResourcePagesIterator;
+import org.apache.beam.sdk.metrics.Counter;
+import org.apache.beam.sdk.metrics.Distribution;
+import org.apache.beam.sdk.metrics.Metrics;
+import org.apache.beam.sdk.schemas.AutoValueSchema;
+import org.apache.beam.sdk.schemas.annotations.DefaultSchema;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionTuple;
+import org.apache.beam.sdk.values.PInput;
+import org.apache.beam.sdk.values.POutput;
+import org.apache.beam.sdk.values.PValue;
+import org.apache.beam.sdk.values.TupleTag;
+import org.apache.beam.sdk.values.TupleTagList;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Throwables;
+import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.collect.ImmutableMap;
+import org.checkerframework.checker.nullness.qual.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/** The type FhirIOPatientEverything for querying a FHIR Patient resource's compartment. * */
+public class FhirIOPatientEverything
+ extends PTransform<PCollection<PatientEverythingParameter>, FhirIOPatientEverything.Result> {
+
+ /** The tag for the main output of FHIR Resources from a GetPatientEverything request. */
+ public static final TupleTag<JsonArray> OUT = new TupleTag<JsonArray>() {};
+ /** The tag for the deadletter output of FHIR Resources from a GetPatientEverything request. */
+ public static final TupleTag<HealthcareIOError<String>> DEAD_LETTER =
+ new TupleTag<HealthcareIOError<String>>() {};
+
+ /**
+ * PatientEverythingParameter defines required attributes for a FHIR GetPatientEverything request
+ * in {@link FhirIOPatientEverything}. *
+ */
+ @DefaultSchema(AutoValueSchema.class)
+ @AutoValue
+ public abstract static class PatientEverythingParameter implements Serializable {
+
+ /**
+ * FHIR Patient resource name in the format of
+ * projects/{p}/locations/{l}/datasets/{d}/fhirStores/{f}/fhir/{resourceType}/{id}.
+ */
+ abstract String getResourceName();
+ /** Optional filters for the request, eg. start, end, _type, _since, _count */
+ abstract @Nullable Map<String, String> getFilters();
+
+ static Builder builder() {
+ return new AutoValue_FhirIOPatientEverything_PatientEverythingParameter.Builder();
+ }
+
+ @AutoValue.Builder
+ abstract static class Builder {
+ abstract Builder setResourceName(String resourceName);
+
+ abstract Builder setFilters(Map<String, String> query);
+
+ abstract PatientEverythingParameter build();
+ }
+ }
+
+ /** The Result for a {@link FhirIOPatientEverything} request. */
+ public static class Result implements POutput, PInput {
+
+ private final PCollection<JsonArray> patientCompartments;
+ private final PCollection<HealthcareIOError<String>> failedReads;
+
+ PCollectionTuple pct;
+
+ /**
+ * Create FhirIOPatientEverything.Result form PCollectionTuple with OUT and DEAD_LETTER tags.
+ *
+ * @param pct the pct
+ * @return the patient everything result
+ * @throws IllegalArgumentException the illegal argument exception
+ */
+ static Result of(PCollectionTuple pct) throws IllegalArgumentException {
+ if (pct.has(OUT) && pct.has(DEAD_LETTER)) {
+ return new Result(pct);
+ } else {
+ throw new IllegalArgumentException(
+ "The PCollection tuple must have the FhirIOPatientEverything.OUT "
+ + "and FhirIOPatientEverything.DEAD_LETTER tuple tags");
+ }
+ }
+
+ private Result(PCollectionTuple pct) {
+ this.pct = pct;
+ this.patientCompartments = pct.get(OUT).setCoder(JsonArrayCoder.of());
+ this.failedReads =
+ pct.get(DEAD_LETTER).setCoder(HealthcareIOErrorCoder.of(StringUtf8Coder.of()));
+ }
+
+ /**
+ * Gets failed reads.
+ *
+ * @return the failed reads
+ */
+ public PCollection<HealthcareIOError<String>> getFailedReads() {
+ return failedReads;
+ }
+
+ /**
+ * Gets the patient compartment responses for GetPatientEverything requests.
+ *
+ * @return the read patient compartments
+ */
+ public PCollection<JsonArray> getPatientCompartments() {
+ return patientCompartments;
+ }
+
+ @Override
+ public Pipeline getPipeline() {
+ return this.pct.getPipeline();
+ }
+
+ @Override
+ public Map<TupleTag<?>, PValue> expand() {
Review comment:
oh my bad! I misread this. Thanks!
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] pabloem merged pull request #15936: [BEAM-13351] FhirIO GetPatientEverything connector
Posted by GitBox <gi...@apache.org>.
pabloem merged pull request #15936:
URL: https://github.com/apache/beam/pull/15936
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on pull request #15936: FhirIO GetPatientEverything connector
Posted by GitBox <gi...@apache.org>.
lukecwik commented on pull request #15936:
URL: https://github.com/apache/beam/pull/15936#issuecomment-966555106
R: @pabloem
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] msbukal commented on a change in pull request #15936: FhirIO GetPatientEverything connector
Posted by GitBox <gi...@apache.org>.
msbukal commented on a change in pull request #15936:
URL: https://github.com/apache/beam/pull/15936#discussion_r756338986
##########
File path: sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/healthcare/FhirIO.java
##########
@@ -1339,13 +1353,29 @@ public void importBatch(FinishBundleContext context) throws IOException {
this.fhirStore = fhirStore;
}
- /**
- * Instantiates a new Execute bundles.
- *
- * @param fhirStore the fhir store
- */
- ExecuteBundles(String fhirStore) {
- this.fhirStore = StaticValueProvider.of(fhirStore);
+ @Override
+ ValueProvider<String> getFhirStore() {
+ return null;
Review comment:
Done
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org