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 2022/07/26 22:47:37 UTC

[GitHub] [beam] lnogueir opened a new pull request, #22460: Add Import transform to Go FhirIO

lnogueir opened a new pull request, #22460:
URL: https://github.com/apache/beam/pull/22460

   Add Import transform to Go FhirIO to enable bulk importing Fhir resources located on a Google Cloud Storage bucket into a provided Google Cloud Fhir Store.
   
   Fixes 22459
   ------------------------
   
   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`).
    - [ ] Mention the appropriate issue in your description (for example: `addresses #123`), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment `fixes #<ISSUE NUMBER>` instead.
    - [ ] 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).
   
   To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
   
   GitHub Actions Tests Status (on master branch)
   ------------------------------------------------------------------------------------------------
   [![Build python source distribution and wheels](https://github.com/apache/beam/workflows/Build%20python%20source%20distribution%20and%20wheels/badge.svg?branch=master&event=schedule)](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
   [![Python tests](https://github.com/apache/beam/workflows/Python%20tests/badge.svg?branch=master&event=schedule)](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
   [![Java tests](https://github.com/apache/beam/workflows/Java%20Tests/badge.svg?branch=master&event=schedule)](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] github-actions[bot] commented on pull request #22460: Add Import transform to Go FhirIO

Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #22460:
URL: https://github.com/apache/beam/pull/22460#issuecomment-1196056277

   Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control


-- 
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] lostluck commented on a diff in pull request #22460: Add Import transform to Go FhirIO

Posted by GitBox <gi...@apache.org>.
lostluck commented on code in PR #22460:
URL: https://github.com/apache/beam/pull/22460#discussion_r938294348


##########
sdks/go/pkg/beam/io/fhirio/import.go:
##########
@@ -0,0 +1,250 @@
+// 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 fhirio provides an API for reading and writing resources to Google
+// Cloud Healthcare Fhir stores.
+// Experimental.
+package fhirio
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"io"
+	"path/filepath"
+	"strings"
+
+	"github.com/apache/beam/sdks/v2/go/pkg/beam"
+	"github.com/apache/beam/sdks/v2/go/pkg/beam/io/filesystem"
+	_ "github.com/apache/beam/sdks/v2/go/pkg/beam/io/filesystem/gcs"
+	_ "github.com/apache/beam/sdks/v2/go/pkg/beam/io/filesystem/local"
+	"github.com/apache/beam/sdks/v2/go/pkg/beam/log"
+	"github.com/apache/beam/sdks/v2/go/pkg/beam/register"
+	"github.com/google/uuid"
+)
+
+func init() {
+	register.DoFn3x0[context.Context, string, func(string)](&importFn{})
+	register.DoFn4x0[context.Context, string, func(string), func(string)](&createBatchFilesFn{})
+	register.Emitter1[string]()
+}
+
+// ContentStructure representation as per:
+// https://cloud.google.com/healthcare-api/docs/reference/rest/v1/projects.locations.datasets.fhirStores/import#contentstructure
+type ContentStructure int
+
+const (
+	ContentStructureUnspecified ContentStructure = iota
+	ContentStructureBundle
+	ContentStructureResource
+)
+
+func (cs ContentStructure) String() string {
+	switch cs {
+	case ContentStructureBundle:
+		return "BUNDLE"
+	case ContentStructureResource:
+		return "RESOURCE"
+	case ContentStructureUnspecified:
+		fallthrough
+	default:
+		return "CONTENT_STRUCTURE_UNSPECIFIED"
+	}
+}
+
+type createBatchFilesFn struct {
+	fs              filesystem.Interface
+	batchFileWriter io.WriteCloser
+	batchFilePath   string
+	TempLocation    string
+}
+
+func (fn *createBatchFilesFn) StartBundle(ctx context.Context, _, _ func(string)) error {
+	fs, err := filesystem.New(ctx, fn.TempLocation)
+	if err != nil {
+		return err
+	}
+	fn.fs = fs
+	fn.batchFilePath = fmt.Sprintf("%s/fhirImportBatch-%v.ndjson", fn.TempLocation, uuid.New())
+	log.Infof(ctx, "Opening to write batch file: %v", fn.batchFilePath)
+	fn.batchFileWriter, err = fn.fs.OpenWrite(ctx, fn.batchFilePath)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (fn *createBatchFilesFn) ProcessElement(ctx context.Context, resource string, _, emitFailedResource func(string)) {
+	_, err := fn.batchFileWriter.Write([]byte(resource + "\n"))
+	if err != nil {
+		log.Warnf(ctx, "Failed to write resource to batch file. Reason: %v", err)
+		emitFailedResource(resource)
+	}
+}
+
+func (fn *createBatchFilesFn) FinishBundle(ctx context.Context, emitBatchFilePath, _ func(string)) {
+	fn.batchFileWriter.Close()
+	fn.batchFileWriter = nil
+	fn.fs.Close()
+	fn.fs = nil
+	emitBatchFilePath(fn.batchFilePath)
+	log.Infof(ctx, "Batch file created: %v", fn.batchFilePath)
+}
+
+type importFn struct {
+	fnCommonVariables
+	operationCounters
+	fs                               filesystem.Interface
+	batchFilesPath                   []string
+	tempBatchDir                     string
+	FhirStorePath                    string
+	TempLocation, DeadLetterLocation string
+	ContentStructure                 ContentStructure
+}
+
+func (fn importFn) String() string {
+	return "importFn"
+}
+
+func (fn *importFn) Setup() {
+	fn.fnCommonVariables.setup(fn.String())
+	fn.operationCounters.setup(fn.String())
+}
+
+func (fn *importFn) StartBundle(ctx context.Context, _ func(string)) error {
+	fs, err := filesystem.New(ctx, fn.TempLocation)
+	if err != nil {
+		return err
+	}
+	fn.fs = fs
+	fn.batchFilesPath = make([]string, 0)
+	fn.tempBatchDir = fmt.Sprintf("%s/tmp-%v", fn.TempLocation, uuid.New())
+	return nil
+}
+
+func (fn *importFn) ProcessElement(ctx context.Context, batchFilePath string, _ func(string)) {
+	updatedBatchFilePath := fmt.Sprintf("%s/%s", fn.tempBatchDir, filepath.Base(batchFilePath))
+	err := filesystem.Rename(ctx, fn.fs, batchFilePath, updatedBatchFilePath)
+	if err != nil {
+		updatedBatchFilePath = batchFilePath
+		log.Warnf(ctx, "Failed to move %v to temp location. Reason: %v", batchFilePath, err)
+	}
+	fn.batchFilesPath = append(fn.batchFilesPath, updatedBatchFilePath)
+}
+
+func (fn *importFn) FinishBundle(ctx context.Context, emitDeadLetter func(string)) {
+	defer func() {
+		fn.fs.Close()
+		fn.fs = nil
+		fn.batchFilesPath = nil
+	}()
+
+	importURI := fn.tempBatchDir + "/*.ndjson"
+	log.Infof(ctx, "About to begin import operation with importURI: %v", importURI)
+	result, err := executeAndRecordLatency(ctx, &fn.latencyMs, func() (operationResults, error) {
+		return fn.client.importResources(fn.FhirStorePath, importURI, fn.ContentStructure)
+	})
+	if err != nil {
+		fn.moveToDeadLetterOrRemoveFailedImportBatchFiles(ctx)
+		fn.operationCounters.errorCount.Inc(ctx, 1)
+		deadLetterMessage := fmt.Sprintf("Failed to import [%v]. Reason: %v", importURI, err)
+		log.Warn(ctx, deadLetterMessage)
+		emitDeadLetter(deadLetterMessage)
+		return
+	}
+
+	log.Infof(ctx, "Imported %v. Results: %v", importURI, result)
+	fn.operationCounters.successCount.Inc(ctx, 1)
+	fn.resourcesSuccessCount.Inc(ctx, result.Successes)
+	fn.resourcesErrorCount.Inc(ctx, result.Failures)
+	fn.removeTempBatchFiles(ctx)
+}
+
+func (fn *importFn) moveToDeadLetterOrRemoveFailedImportBatchFiles(ctx context.Context) {
+	if fn.DeadLetterLocation == "" {
+		log.Info(ctx, "Deadletter path not provided. Remove failed import batch files instead.")
+		fn.removeTempBatchFiles(ctx)
+		return
+	}
+
+	log.Infof(ctx, "Moving failed import files to Deadletter path: [%v]", fn.DeadLetterLocation)
+	for _, p := range fn.batchFilesPath {
+		err := filesystem.Rename(ctx, fn.fs, p, fmt.Sprintf("%s/%s", fn.DeadLetterLocation, filepath.Base(p)))
+		if err != nil {
+			log.Warnf(ctx, "Failed to move failed imported file %v to %v. Reason: %v", p, fn.DeadLetterLocation, err)
+		}
+	}
+}
+
+func (fn *importFn) removeTempBatchFiles(ctx context.Context) {
+	for _, p := range fn.batchFilesPath {
+		err := fn.fs.(filesystem.Remover).Remove(ctx, p)
+		if err != nil {
+			log.Warnf(ctx, "Failed to delete temp batch file [%v]. Reason: %v", p, err)
+		}
+	}
+}
+
+// Import consumes FHIR resources as input PCollection<string> and imports them
+// into a given Google Cloud Healthcare FHIR store. It does so by creating batch
+// files in the provided Google Cloud Storage `tempDir` and importing those files
+// to the store through FHIR import API method: https://cloud.google.com/healthcare-api/docs/concepts/fhir-import.
+// If `tempDir` is not provided, it falls back to the dataflow temp_location flag.
+// Resources that fail to be included in the batch files are included as the
+// first output PCollection. In case a batch file fails to be imported, it will
+// be moved to the `deadLetterDir` and an error message will be provided in the
+// second output PCollection. If `deadLetterDir` is not provided, the failed
+// import files will be deleted and be irretrievable, but the error message will
+// still be provided.
+func Import(s beam.Scope, fhirStorePath, tempDir, deadLetterDir string, contentStructure ContentStructure, resources beam.PCollection) (beam.PCollection, beam.PCollection) {
+	s = s.Scope("fhirio.Import")
+
+	if tempDir == "" {
+		tempDir = tryFallbackToDataflowTempDirOrPanic()
+	}
+	tempDir = strings.TrimSuffix(tempDir, "/")
+	deadLetterDir = strings.TrimSuffix(deadLetterDir, "/")
+
+	return importResourcesInBatches(s, fhirStorePath, tempDir, deadLetterDir, contentStructure, resources, nil)
+}
+
+// This is useful as an entry point for testing because we can provide a fake FHIR store client.
+func importResourcesInBatches(s beam.Scope, fhirStorePath, tempDir, deadLetterDir string, contentStructure ContentStructure, resources beam.PCollection, client fhirStoreClient) (beam.PCollection, beam.PCollection) {
+	batchFiles, failedResources := beam.ParDo2(s, &createBatchFilesFn{TempLocation: tempDir}, resources)
+	failedImportsDeadLetter := beam.ParDo(
+		s,
+		&importFn{
+			fnCommonVariables:  fnCommonVariables{client: client},
+			FhirStorePath:      fhirStorePath,
+			TempLocation:       tempDir,
+			DeadLetterLocation: deadLetterDir,
+			ContentStructure:   contentStructure,
+		},
+		batchFiles,
+	)
+	return failedResources, failedImportsDeadLetter
+}
+
+func tryFallbackToDataflowTempDirOrPanic() string {
+	if f := flag.Lookup("temp_location"); f != nil && f.Value.String() != "" {

Review Comment:
   Technically these would need to be looked up via the "beam.PipelineOptions" lookup on workers, as I don't think these flags are populated on the worker process.
   
   I'd leave it as is for now, and put a TODO, unless you're keen on making the update:
   Here's an example:
   https://github.com/apache/beam/blob/67e6726ffeb47d2ada0122369fa230833ce0f026/sdks/go/test/integration/primitives/pardo.go#L190



-- 
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] lostluck commented on a diff in pull request #22460: Add Import transform to Go FhirIO

Posted by GitBox <gi...@apache.org>.
lostluck commented on code in PR #22460:
URL: https://github.com/apache/beam/pull/22460#discussion_r938292851


##########
sdks/go/pkg/beam/io/fhirio/import.go:
##########
@@ -0,0 +1,247 @@
+// 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 fhirio provides an API for reading and writing resources to Google
+// Cloud Healthcare Fhir stores.
+// Experimental.
+package fhirio
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"io"
+	"path/filepath"
+	"strings"
+
+	"github.com/apache/beam/sdks/v2/go/pkg/beam"
+	"github.com/apache/beam/sdks/v2/go/pkg/beam/io/filesystem"
+	_ "github.com/apache/beam/sdks/v2/go/pkg/beam/io/filesystem/gcs"
+	_ "github.com/apache/beam/sdks/v2/go/pkg/beam/io/filesystem/local"
+	"github.com/apache/beam/sdks/v2/go/pkg/beam/log"
+	"github.com/apache/beam/sdks/v2/go/pkg/beam/register"
+	"github.com/google/uuid"
+)
+
+func init() {
+	register.DoFn3x0[context.Context, string, func(string)]((*importFn)(nil))

Review Comment:
   FYI what was there was correct, and arguably preferred.
   
   The original is a zero allocation typed nil, while the new code will allocate the DoFn struct (which will then get garbage collected relatively quickly anyway).
   
   Too late to change it now, but ideally we'd have put the type as part of the generic types instead of passing in as an argument, but I think we're praying for better type inference in the future...



-- 
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] lnogueir commented on a diff in pull request #22460: Add Import transform to Go FhirIO

Posted by GitBox <gi...@apache.org>.
lnogueir commented on code in PR #22460:
URL: https://github.com/apache/beam/pull/22460#discussion_r938933341


##########
sdks/go/pkg/beam/io/fhirio/import.go:
##########
@@ -0,0 +1,250 @@
+// 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 fhirio provides an API for reading and writing resources to Google
+// Cloud Healthcare Fhir stores.
+// Experimental.
+package fhirio
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"io"
+	"path/filepath"
+	"strings"
+
+	"github.com/apache/beam/sdks/v2/go/pkg/beam"
+	"github.com/apache/beam/sdks/v2/go/pkg/beam/io/filesystem"
+	_ "github.com/apache/beam/sdks/v2/go/pkg/beam/io/filesystem/gcs"
+	_ "github.com/apache/beam/sdks/v2/go/pkg/beam/io/filesystem/local"
+	"github.com/apache/beam/sdks/v2/go/pkg/beam/log"
+	"github.com/apache/beam/sdks/v2/go/pkg/beam/register"
+	"github.com/google/uuid"
+)
+
+func init() {
+	register.DoFn3x0[context.Context, string, func(string)](&importFn{})
+	register.DoFn4x0[context.Context, string, func(string), func(string)](&createBatchFilesFn{})
+	register.Emitter1[string]()
+}
+
+// ContentStructure representation as per:
+// https://cloud.google.com/healthcare-api/docs/reference/rest/v1/projects.locations.datasets.fhirStores/import#contentstructure
+type ContentStructure int
+
+const (
+	ContentStructureUnspecified ContentStructure = iota
+	ContentStructureBundle
+	ContentStructureResource
+)
+
+func (cs ContentStructure) String() string {
+	switch cs {
+	case ContentStructureBundle:
+		return "BUNDLE"
+	case ContentStructureResource:
+		return "RESOURCE"
+	case ContentStructureUnspecified:
+		fallthrough
+	default:
+		return "CONTENT_STRUCTURE_UNSPECIFIED"
+	}
+}
+
+type createBatchFilesFn struct {
+	fs              filesystem.Interface
+	batchFileWriter io.WriteCloser
+	batchFilePath   string
+	TempLocation    string
+}
+
+func (fn *createBatchFilesFn) StartBundle(ctx context.Context, _, _ func(string)) error {
+	fs, err := filesystem.New(ctx, fn.TempLocation)
+	if err != nil {
+		return err
+	}
+	fn.fs = fs
+	fn.batchFilePath = fmt.Sprintf("%s/fhirImportBatch-%v.ndjson", fn.TempLocation, uuid.New())
+	log.Infof(ctx, "Opening to write batch file: %v", fn.batchFilePath)
+	fn.batchFileWriter, err = fn.fs.OpenWrite(ctx, fn.batchFilePath)
+	if err != nil {
+		return err
+	}
+	return nil
+}
+
+func (fn *createBatchFilesFn) ProcessElement(ctx context.Context, resource string, _, emitFailedResource func(string)) {
+	_, err := fn.batchFileWriter.Write([]byte(resource + "\n"))
+	if err != nil {
+		log.Warnf(ctx, "Failed to write resource to batch file. Reason: %v", err)
+		emitFailedResource(resource)
+	}
+}
+
+func (fn *createBatchFilesFn) FinishBundle(ctx context.Context, emitBatchFilePath, _ func(string)) {
+	fn.batchFileWriter.Close()
+	fn.batchFileWriter = nil
+	fn.fs.Close()
+	fn.fs = nil
+	emitBatchFilePath(fn.batchFilePath)
+	log.Infof(ctx, "Batch file created: %v", fn.batchFilePath)
+}
+
+type importFn struct {
+	fnCommonVariables
+	operationCounters
+	fs                               filesystem.Interface
+	batchFilesPath                   []string
+	tempBatchDir                     string
+	FhirStorePath                    string
+	TempLocation, DeadLetterLocation string
+	ContentStructure                 ContentStructure
+}
+
+func (fn importFn) String() string {
+	return "importFn"
+}
+
+func (fn *importFn) Setup() {
+	fn.fnCommonVariables.setup(fn.String())
+	fn.operationCounters.setup(fn.String())
+}
+
+func (fn *importFn) StartBundle(ctx context.Context, _ func(string)) error {
+	fs, err := filesystem.New(ctx, fn.TempLocation)
+	if err != nil {
+		return err
+	}
+	fn.fs = fs
+	fn.batchFilesPath = make([]string, 0)
+	fn.tempBatchDir = fmt.Sprintf("%s/tmp-%v", fn.TempLocation, uuid.New())
+	return nil
+}
+
+func (fn *importFn) ProcessElement(ctx context.Context, batchFilePath string, _ func(string)) {
+	updatedBatchFilePath := fmt.Sprintf("%s/%s", fn.tempBatchDir, filepath.Base(batchFilePath))
+	err := filesystem.Rename(ctx, fn.fs, batchFilePath, updatedBatchFilePath)
+	if err != nil {
+		updatedBatchFilePath = batchFilePath
+		log.Warnf(ctx, "Failed to move %v to temp location. Reason: %v", batchFilePath, err)
+	}
+	fn.batchFilesPath = append(fn.batchFilesPath, updatedBatchFilePath)
+}
+
+func (fn *importFn) FinishBundle(ctx context.Context, emitDeadLetter func(string)) {
+	defer func() {
+		fn.fs.Close()
+		fn.fs = nil
+		fn.batchFilesPath = nil
+	}()
+
+	importURI := fn.tempBatchDir + "/*.ndjson"
+	log.Infof(ctx, "About to begin import operation with importURI: %v", importURI)
+	result, err := executeAndRecordLatency(ctx, &fn.latencyMs, func() (operationResults, error) {
+		return fn.client.importResources(fn.FhirStorePath, importURI, fn.ContentStructure)
+	})
+	if err != nil {
+		fn.moveToDeadLetterOrRemoveFailedImportBatchFiles(ctx)
+		fn.operationCounters.errorCount.Inc(ctx, 1)
+		deadLetterMessage := fmt.Sprintf("Failed to import [%v]. Reason: %v", importURI, err)
+		log.Warn(ctx, deadLetterMessage)
+		emitDeadLetter(deadLetterMessage)
+		return
+	}
+
+	log.Infof(ctx, "Imported %v. Results: %v", importURI, result)
+	fn.operationCounters.successCount.Inc(ctx, 1)
+	fn.resourcesSuccessCount.Inc(ctx, result.Successes)
+	fn.resourcesErrorCount.Inc(ctx, result.Failures)
+	fn.removeTempBatchFiles(ctx)
+}
+
+func (fn *importFn) moveToDeadLetterOrRemoveFailedImportBatchFiles(ctx context.Context) {
+	if fn.DeadLetterLocation == "" {
+		log.Info(ctx, "Deadletter path not provided. Remove failed import batch files instead.")
+		fn.removeTempBatchFiles(ctx)
+		return
+	}
+
+	log.Infof(ctx, "Moving failed import files to Deadletter path: [%v]", fn.DeadLetterLocation)
+	for _, p := range fn.batchFilesPath {
+		err := filesystem.Rename(ctx, fn.fs, p, fmt.Sprintf("%s/%s", fn.DeadLetterLocation, filepath.Base(p)))
+		if err != nil {
+			log.Warnf(ctx, "Failed to move failed imported file %v to %v. Reason: %v", p, fn.DeadLetterLocation, err)
+		}
+	}
+}
+
+func (fn *importFn) removeTempBatchFiles(ctx context.Context) {
+	for _, p := range fn.batchFilesPath {
+		err := fn.fs.(filesystem.Remover).Remove(ctx, p)
+		if err != nil {
+			log.Warnf(ctx, "Failed to delete temp batch file [%v]. Reason: %v", p, err)
+		}
+	}
+}
+
+// Import consumes FHIR resources as input PCollection<string> and imports them
+// into a given Google Cloud Healthcare FHIR store. It does so by creating batch
+// files in the provided Google Cloud Storage `tempDir` and importing those files
+// to the store through FHIR import API method: https://cloud.google.com/healthcare-api/docs/concepts/fhir-import.
+// If `tempDir` is not provided, it falls back to the dataflow temp_location flag.
+// Resources that fail to be included in the batch files are included as the
+// first output PCollection. In case a batch file fails to be imported, it will
+// be moved to the `deadLetterDir` and an error message will be provided in the
+// second output PCollection. If `deadLetterDir` is not provided, the failed
+// import files will be deleted and be irretrievable, but the error message will
+// still be provided.
+func Import(s beam.Scope, fhirStorePath, tempDir, deadLetterDir string, contentStructure ContentStructure, resources beam.PCollection) (beam.PCollection, beam.PCollection) {
+	s = s.Scope("fhirio.Import")
+
+	if tempDir == "" {
+		tempDir = tryFallbackToDataflowTempDirOrPanic()
+	}
+	tempDir = strings.TrimSuffix(tempDir, "/")
+	deadLetterDir = strings.TrimSuffix(deadLetterDir, "/")
+
+	return importResourcesInBatches(s, fhirStorePath, tempDir, deadLetterDir, contentStructure, resources, nil)
+}
+
+// This is useful as an entry point for testing because we can provide a fake FHIR store client.
+func importResourcesInBatches(s beam.Scope, fhirStorePath, tempDir, deadLetterDir string, contentStructure ContentStructure, resources beam.PCollection, client fhirStoreClient) (beam.PCollection, beam.PCollection) {
+	batchFiles, failedResources := beam.ParDo2(s, &createBatchFilesFn{TempLocation: tempDir}, resources)
+	failedImportsDeadLetter := beam.ParDo(
+		s,
+		&importFn{
+			fnCommonVariables:  fnCommonVariables{client: client},
+			FhirStorePath:      fhirStorePath,
+			TempLocation:       tempDir,
+			DeadLetterLocation: deadLetterDir,
+			ContentStructure:   contentStructure,
+		},
+		batchFiles,
+	)
+	return failedResources, failedImportsDeadLetter
+}
+
+func tryFallbackToDataflowTempDirOrPanic() string {
+	if f := flag.Lookup("temp_location"); f != nil && f.Value.String() != "" {

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

[GitHub] [beam] jrmccluskey commented on pull request #22460: Add Import transform to Go FhirIO

Posted by GitBox <gi...@apache.org>.
jrmccluskey commented on PR #22460:
URL: https://github.com/apache/beam/pull/22460#issuecomment-1199935721

   Run Go 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] lnogueir commented on a diff in pull request #22460: Add Import transform to Go FhirIO

Posted by GitBox <gi...@apache.org>.
lnogueir commented on code in PR #22460:
URL: https://github.com/apache/beam/pull/22460#discussion_r938934270


##########
sdks/go/pkg/beam/io/fhirio/import.go:
##########
@@ -0,0 +1,247 @@
+// 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 fhirio provides an API for reading and writing resources to Google
+// Cloud Healthcare Fhir stores.
+// Experimental.
+package fhirio
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"io"
+	"path/filepath"
+	"strings"
+
+	"github.com/apache/beam/sdks/v2/go/pkg/beam"
+	"github.com/apache/beam/sdks/v2/go/pkg/beam/io/filesystem"
+	_ "github.com/apache/beam/sdks/v2/go/pkg/beam/io/filesystem/gcs"
+	_ "github.com/apache/beam/sdks/v2/go/pkg/beam/io/filesystem/local"
+	"github.com/apache/beam/sdks/v2/go/pkg/beam/log"
+	"github.com/apache/beam/sdks/v2/go/pkg/beam/register"
+	"github.com/google/uuid"
+)
+
+func init() {
+	register.DoFn3x0[context.Context, string, func(string)]((*importFn)(nil))

Review Comment:
   Cool! Reverted the changes back to how they were originally.



-- 
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] riteshghorse commented on a diff in pull request #22460: Add Import transform to Go FhirIO

Posted by GitBox <gi...@apache.org>.
riteshghorse commented on code in PR #22460:
URL: https://github.com/apache/beam/pull/22460#discussion_r938178966


##########
sdks/go/pkg/beam/io/fhirio/import.go:
##########
@@ -0,0 +1,250 @@
+// 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 fhirio provides an API for reading and writing resources to Google
+// Cloud Healthcare Fhir stores.
+// Experimental.
+package fhirio
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"io"
+	"path/filepath"
+	"strings"
+
+	"github.com/apache/beam/sdks/v2/go/pkg/beam"
+	"github.com/apache/beam/sdks/v2/go/pkg/beam/io/filesystem"
+	_ "github.com/apache/beam/sdks/v2/go/pkg/beam/io/filesystem/gcs"
+	_ "github.com/apache/beam/sdks/v2/go/pkg/beam/io/filesystem/local"
+	"github.com/apache/beam/sdks/v2/go/pkg/beam/log"
+	"github.com/apache/beam/sdks/v2/go/pkg/beam/register"
+	"github.com/google/uuid"
+)
+
+func init() {
+	register.DoFn3x0[context.Context, string, func(string)](&importFn{})
+	register.DoFn4x0[context.Context, string, func(string), func(string)](&createBatchFilesFn{})
+	register.Emitter1[string]()
+}
+
+// ContentStructure representation as per:
+// https://cloud.google.com/healthcare-api/docs/reference/rest/v1/projects.locations.datasets.fhirStores/import#contentstructure
+type ContentStructure int
+
+const (
+	ContentStructureUnspecified ContentStructure = iota

Review Comment:
   Add comment describing the constant's use. These may create lint errors on import.



-- 
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] lostluck merged pull request #22460: Add Import transform to Go FhirIO

Posted by GitBox <gi...@apache.org>.
lostluck merged PR #22460:
URL: https://github.com/apache/beam/pull/22460


-- 
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] codecov[bot] commented on pull request #22460: Add Import transform to Go FhirIO

Posted by GitBox <gi...@apache.org>.
codecov[bot] commented on PR #22460:
URL: https://github.com/apache/beam/pull/22460#issuecomment-1196059799

   # [Codecov](https://codecov.io/gh/apache/beam/pull/22460?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
   > Merging [#22460](https://codecov.io/gh/apache/beam/pull/22460?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (b7b7ef2) into [master](https://codecov.io/gh/apache/beam/commit/4b5efc3db490b4cc6a017a917cde561ace03f9b1?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (4b5efc3) will **not change** coverage.
   > The diff coverage is `21.42%`.
   
   ```diff
   @@           Coverage Diff           @@
   ##           master   #22460   +/-   ##
   =======================================
     Coverage   74.16%   74.16%           
   =======================================
     Files         706      706           
     Lines       93195    93195           
   =======================================
     Hits        69122    69122           
     Misses      22805    22805           
     Partials     1268     1268           
   ```
   
   
   | [Impacted Files](https://codecov.io/gh/apache/beam/pull/22460?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
   |---|---|---|
   | [sdks/go/pkg/beam/io/fhirio/common.go](https://codecov.io/gh/apache/beam/pull/22460/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9pby9maGlyaW8vY29tbW9uLmdv) | `36.58% <0.00%> (ø)` | |
   | [sdks/go/pkg/beam/io/fhirio/deidentify.go](https://codecov.io/gh/apache/beam/pull/22460/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9pby9maGlyaW8vZGVpZGVudGlmeS5nbw==) | `91.89% <100.00%> (ø)` | |
   
   Help us with your feedback. Take ten seconds to tell us [how you rate us](https://about.codecov.io/nps?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
   


-- 
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] jrmccluskey commented on a diff in pull request #22460: Add Import transform to Go FhirIO

Posted by GitBox <gi...@apache.org>.
jrmccluskey commented on code in PR #22460:
URL: https://github.com/apache/beam/pull/22460#discussion_r933596239


##########
sdks/go/pkg/beam/io/fhirio/import.go:
##########
@@ -0,0 +1,247 @@
+// 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 fhirio provides an API for reading and writing resources to Google
+// Cloud Healthcare Fhir stores.
+// Experimental.
+package fhirio
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"io"
+	"path/filepath"
+	"strings"
+
+	"github.com/apache/beam/sdks/v2/go/pkg/beam"
+	"github.com/apache/beam/sdks/v2/go/pkg/beam/io/filesystem"
+	_ "github.com/apache/beam/sdks/v2/go/pkg/beam/io/filesystem/gcs"
+	_ "github.com/apache/beam/sdks/v2/go/pkg/beam/io/filesystem/local"
+	"github.com/apache/beam/sdks/v2/go/pkg/beam/log"
+	"github.com/apache/beam/sdks/v2/go/pkg/beam/register"
+	"github.com/google/uuid"
+)
+
+func init() {
+	register.DoFn3x0[context.Context, string, func(string)]((*importFn)(nil))
+	register.DoFn4x0[context.Context, string, func(string), func(string)]((*createBatchFilesFn)(nil))

Review Comment:
   ```suggestion
   	register.DoFn4x0[context.Context, string, func(string), func(string)]((&createBatchFilesFn{})(nil))
   ```



##########
sdks/go/pkg/beam/io/fhirio/import.go:
##########
@@ -0,0 +1,247 @@
+// 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 fhirio provides an API for reading and writing resources to Google
+// Cloud Healthcare Fhir stores.
+// Experimental.
+package fhirio
+
+import (
+	"context"
+	"flag"
+	"fmt"
+	"io"
+	"path/filepath"
+	"strings"
+
+	"github.com/apache/beam/sdks/v2/go/pkg/beam"
+	"github.com/apache/beam/sdks/v2/go/pkg/beam/io/filesystem"
+	_ "github.com/apache/beam/sdks/v2/go/pkg/beam/io/filesystem/gcs"
+	_ "github.com/apache/beam/sdks/v2/go/pkg/beam/io/filesystem/local"
+	"github.com/apache/beam/sdks/v2/go/pkg/beam/log"
+	"github.com/apache/beam/sdks/v2/go/pkg/beam/register"
+	"github.com/google/uuid"
+)
+
+func init() {
+	register.DoFn3x0[context.Context, string, func(string)]((*importFn)(nil))

Review Comment:
   Registration here should be 
   ```suggestion
   	register.DoFn3x0[context.Context, string, func(string)]((&importFn{})(nil))
   ```



-- 
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] lnogueir commented on pull request #22460: Add Import transform to Go FhirIO

Posted by GitBox <gi...@apache.org>.
lnogueir commented on PR #22460:
URL: https://github.com/apache/beam/pull/22460#issuecomment-1196055581

   R: @msbukal 


-- 
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] jrmccluskey commented on pull request #22460: Add Import transform to Go FhirIO

Posted by GitBox <gi...@apache.org>.
jrmccluskey commented on PR #22460:
URL: https://github.com/apache/beam/pull/22460#issuecomment-1201584284

   Run Go 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