The avro definition is included in the files the main union type is organised as followed, the osmtype mention which member that will be populated.
Points are modeled using x/y, avoiding to use a geometry library to consume. This is very handy in practice.All coordinates are provided in the avro file using WGS84 Coordinate system.
The full Avro schema definition is provided here
Complex geometry entities (ways, polygons) are exposed in byte array in the schema. A library must be then use for getting a Geometry object from the byte array.
The geometry are encoded in binary shape format, this format is fully described in this paper if you wish to decode, https://www.esri.com/library/whitepapers/pdfs/shapefile.pdf
ESRI Offer for JVM, a geometry api, permitting to load, save, and work with geometries as standalone library : https://github.com/Esri/geometry-api-java.
It Provides:
The following example use the net.frett27:osm-gis-avro:0.1 dependency.
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
AvroInputFormat p = new AvroInputFormat<>(new Path("c:\\temp\\1.avro"), OSMEntity.class);
DataSource inputDataset = env.createInput(p);
System.out.println(inputDataset.count());
associated maven dependencies
compile 'com.esri.geometry:esri-geometry-api:1.2.1'
compile "org.apache.flink:flink-java:${flinkversion}"
compile "org.apache.flink:flink-clients_2.10:${flinkversion}"
compile "org.apache.flink:flink-avro_2.10:${flinkversion}"
compile "net.frett27:osm-gis-avro:0.1"
The following example show how to consume the avro stream, thank's to databriks extension
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
/**
* Simple Class for reading avro format from Spark
* @author pfreydiere
*
*/
public class HelloReadOSMSql {
public static void main(String[] args) throws Exception {
// bunch of sys config to make it work on my win laptop
System.setProperty("hadoop.home.dir", "C:\\projets\\2017_spark_avro\\win32");
System.out.println("Read the avro files");
// creating the spark session
SparkSession spark = SparkSession.builder().master("local").getOrCreate();
// Creates a DataFrame from a specified file
Dataset df =
spark.read().format("com.databricks.spark.avro")
.load("C:\\Users\\use\\Downloads\\1.avro");
// only interested in ways
Dataset f = df.filter("osmtype = 'WAY'");
// print out the schema
System.out.println(f.schema());
// get first row
Row r = f.first();
// take the way geometry
Row way = r.getAs("way");
assert way != null;
// geometry is shape binary encoded
byte[] content = way.getAs("geometry");
System.out.println(content);
Geometry g = GeometryEngine.geometryFromEsriShape(content, Type.Polyline);
// dump geojson geometry
System.out.println(GeometryEngine.geometryToGeoJson(g));
}
}
the output result show : {"type":"LineString","coordinates":[[1.5607196,50.3947346],[1.5606827,50.3947002],[1.560464,50.3944957]]}
associated maven dependencies
compile "org.apache.spark:spark-core_2.11:2.2.0"
// https://mvnrepository.com/artifact/org.apache.spark/spark-sql_2.10
compile group: 'org.apache.spark', name: 'spark-sql_2.11', version: '2.2.0'
compile "com.databricks:spark-avro_2.11:3.2.0"