How to Convert PCollection to List Collection in Apache Beam using Java?

I am using Apache beam to process the batch data, for that purpose I am creating PCollection object from List, however once I finish executing pipeline processing, I need to convert the results in PCollection back into List collection. Can anybody please help me with this?

PipelineOptions options = PipelineOptionsFactory.create();
Pipeline pipeline = Pipeline.create(options);

List<Map<String,String>> inputDataList = ...//API call results.

PCollection pcollection = pipeline.apply(Create.of(inputDataList).withCoder(MapCoder.of(StringUtf8Coder.of(),StringUtf8Coder.of())));

pcollection = pcollection.apply()// applying few transformation.

List<Map<String,String>> outputDataList = ? //How to get list from pcollection object?

Did not find much help on internet, whatever solutions mentioned has methods witch are undefined for PCollection. for e.g. get(), GetAllOut() etc.

It is not possible to convert PCollection into a Collection, since Apache Beam is just a logical plan of the transformations and actions but the execution is actually done by other engines like Spark.

What you can do is to let the PCollection to write the results to a Sink (could be a file, GCS, BigQuery, ..) and then you read the results out of this sink.

Checkout the execution model of Apache Beam

Actually, I just encountered this issue. Contrary to the answer and comments above, it is possible to write a PCollection to a Java List, although there are several constraints on doing so, as I discuss below. As far as why you would want to, there are several situations in which it is not adequate to work with PCollections:

1. You want to work with a legacy API, especially one in which the classes are not Java Serializable. Serializability is a requirement of creating transformations on Beam collections.
2. You want to write the contents of a PCollection to some destination other than files, databases, or one of the services that have built-in IO connectors. In my case, I want to write to Java OutputStreams, the common denominator of files, urls, and Swing components such as JTextArea; OutputStream and its subclasses are not Serializable. While Beam can do sophisticated things like handling streaming data, it is surprisingly hard to do things simple like writing text to a screen window.

In order to write a PCollection to a Java List (or other Java Collections), you can use a Beam Combiner. I my case, I use the ArrayAggArray combiner. Writing the data is a two stage process. First you apply the Combiner and run the Pipeline. Then you process the resulting Java object outside the Pipeline. That's the first constraint of this method. Once you've run the Pipeline, you no longer have access to any of the PCollections it has created, so essentially any post-processing must not call back any of the Beam objects.

Note that the Java object you create can be very large and can't be processed in parallel (that's why you want to use Beam in the first place!). That is the second constraint of my method.

The trick is that, in extending the Beam Combiner, you have to save the final accumulator to a Java object outside the Pipeline. I do this by passing a reference to the object in the constructor of the Combiner. In the extractOutput method of the Combiner, you have to add the elements of the final accumulator to the Java object (you can't simply replace it, because Java won't pass a new instance back to the caller).

Here is an example:
First, extend the Beam Combiner:

	public static class JsonWriter extends ArrayAgg.ArrayAggArray<String> {

	private static final long serialVersionUID = -5573525980983338238L;
	
	/**
	 * Holds the JSON serialization of the records in applicable PCollection.
	 * This field serves as the output of the Combine operation on this PCollection. 
	 * It should be passed as an empty list to the constructor, and is populated by the 
	 * extractOutput method.
	 */
	private List<String> records;

	/**
	 * @param records the output of the Combine operation on this PCollection;  
	 * should be passed as an empty list to the constructor
	 * 
	 */
	public JsonWriter(List<String> records) {
		super();
		if(records== null || records.size()!=0) 
			throw new IllegalArgumentException("Records should be empty");
		this.records= records;
	}

	@Override
	public List<String> addInput(
			List<String> accum,
			String input) {
		return super.addInput(accum, input);
	}

	@Override
	public List<String> extractOutput(
			List<String> accumulator) {
		if (accumulator.isEmpty()) {
			return null;
		}
		this.records.addAll(accumulator);
		return accumulator;
	}

	/**
	 * @return the records
	 */
	public List<String> getRecords() {
		return records;
	}

}//class JsonWriter

The caller looks like this:

		instance.recordsToJson(); //instance has a PCollection
		pipeline.run().waitUntilFinish();

where

recordsAsJson = new LinkedList<String>();
/**
 * Maps the records in dataframe to their JSON representation 
 * and collects the resulting strings into the recordsAsJson list.
 * Called as part of the Beam pipeline.
 */
public void recordsToJson() {
	JsonWriter writer= new JsonWriter(recordsAsJson);
	this.dataframe //a PCollection<Row>
		//transform each record into a JSON string
		.apply(
			MapElements.into(TypeDescriptors.strings())
				.via((Row record) -> MsdxInstance.recordToJson(record)))
		//Collect the JSON strings into a list
		.apply(
			this.dataframe.getName() + "AsJson", 			
			Combine.globally(writer));
	
}//recordsToJson

Then you use the list outside the pipeline like this:

/**
 * Writes the JSON serialization of this dataframe to the destination specified by the generator. 
 * Called after the recordsToJson method and after the Beam pipeline runs.
 */
@Override
public MsdxWriter.Generator toJson(MsdxWriter.Generator generator) {
	if(this.dataframe.isBounded().equals(PCollection.IsBounded.UNBOUNDED))
		throw new IllegalArgumentException("Unbounded collection of records");	
	AtomicBoolean firstRecord= new AtomicBoolean(true);
	try {
		generator.writeStartArray(false);
		recordsAsJson.forEach(record -> {			
			try {
				if(!firstRecord.get())
					generator.writeArrayValueSeparator(false);
				else
					firstRecord.set(false);
				generator.writeRaw(record);
			} catch (IOException e) {
				System.err.println(e.getMessage());
				e.printStackTrace();
			}
		});
		generator.writeEndArray(false);
	} catch (IOException e) {
		System.err.println(e.getMessage());
		e.printStackTrace();
	}
	return generator;
}//toJson

And

			window.println("cities:");
		instance.toJson(MsdxOutputDestination.toStream(window.printStream()));

I hope this helps.