add support for openstack swift filesystem

Author: Krzysztof Nagrodkiewicz

examples/swift_write.go

@@ -0,0 +1,104 @@
+package main
+
+import (
+	"fmt"
+	"log"
+	"time"
+
+	"github.com/ncw/swift"
+	"github.com/xitongsys/parquet-go/reader"
+	"github.com/xitongsys/parquet-go/writer"
+
+	"github.com/xitongsys/parquet-go-source/swift"
+)
+
+type Student struct {
+	Name   string  `parquet:"name=name, type=UTF8"`
+	Age    int32   `parquet:"name=age, type=INT32"`
+	Id     int64   `parquet:"name=id, type=INT64"`
+	Weight float32 `parquet:"name=weight, type=FLOAT"`
+	Sex    bool    `parquet:"name=sex, type=BOOLEAN"`
+	Day    int32   `parquet:"name=day, type=DATE"`
+}
+
+func main() {
+	connection := swift.Connection{
+		UserName: "test_user",
+		ApiKey:   "passw0rd",
+		AuthUrl:  "http://localhost:35357/v3",
+		Domain:   "Default",
+		Tenant:   "test_project",
+	}
+	if err := connection.Authenticate(); err != nil {
+		log.Print("Failed to authenticate to keystone: ", err)
+		return
+	}
+
+	containerName := "swift-parquet-test"
+	fileName := "flat.parquet"
+	num := 100
+
+	log.Println("Write started")
+	fw, err := swiftsource.NewSwiftFileWriter(containerName, fileName, &connection)
+	if err != nil {
+		log.Println("Failed to create file: ", err)
+		return
+	}
+
+	//write
+	pw, err := writer.NewParquetWriter(fw, new(Student), 4)
+	if err != nil {
+		log.Println("Failed to create parquet writer: ", err)
+		return
+	}
+
+	for i := 0; i < num; i++ {
+		stu := Student{
+			Name:   fmt.Sprintf("Student-%d", i),
+			Age:    int32(20 + i%5),
+			Id:     int64(i),
+			Weight: 50.0 + float32(i)*0.1,
+			Sex:    i%2 == 0,
+			Day:    int32(time.Now().Unix() / 3600 / 24),
+		}
+		if err = pw.Write(stu); err != nil {
+			log.Println("Write error: ", err)
+		}
+	}
+	if err = pw.WriteStop(); err != nil {
+		log.Println("WriteStop error: ", err)
+		return
+	}
+
+	if err = fw.Close(); err != nil {
+		log.Println("Failed to close writer: ", err)
+	}
+	log.Println("Write finished")
+
+	///read
+	log.Println("Read started")
+	fr, err := swiftsource.NewSwiftFileReader(containerName, fileName, &connection)
+	if err != nil {
+		log.Println("Failed to open file: ", err)
+		return
+	}
+
+	pr, err := reader.NewParquetReader(fr, new(Student), 4)
+	if err != nil {
+		log.Println("Failed to create parquet reader: ", err)
+		return
+	}
+	num = int(pr.GetNumRows())
+	for i := 0; i < num/10; i++ {
+		stus := make([]Student, 10) //read 10 rows
+		if err = pr.Read(&stus); err != nil {
+			log.Println("Read error: ", err)
+		}
+		log.Println(stus)
+	}
+	pr.ReadStop()
+	if err = fr.Close(); err != nil {
+		log.Println("Failed to close reader: ", err)
+	}
+	log.Println("Read finished")
+}

go.mod

@@ -9,6 +9,7 @@ require (
 	github.com/aws/aws-sdk-go v1.30.19
 	github.com/colinmarc/hdfs/v2 v2.1.1
 	github.com/golang/mock v1.4.3
+	github.com/ncw/swift v1.0.52
 	github.com/spf13/afero v1.2.2
 	github.com/xitongsys/parquet-go v1.5.1
 )

go.sum

@@ -132,6 +132,8 @@ github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/mattn/go-ieproxy v0.0.0-20190702010315-6dee0af9227d/go.mod h1:31jz6HNzdxOmlERGGEc4v/dMssOfmp2p5bT/okiKFFc=
 github.com/mattn/go-ieproxy v0.0.1 h1:qiyop7gCflfhwCzGyeT0gro3sF9AIg9HU98JORTkqfI=
 github.com/mattn/go-ieproxy v0.0.1/go.mod h1:pYabZ6IHcRpFh7vIaLfK7rdcWgFEb3SFJ6/gNWuh88E=
+github.com/ncw/swift v1.0.52 h1:ACF3JufDGgeKp/9mrDgQlEgS8kRYC4XKcuzj/8EJjQU=
+github.com/ncw/swift v1.0.52/go.mod h1:23YIA4yWVnGwv2dQlN4bB7egfYX6YLn0Yo/S6zZO/ZM=
 github.com/pborman/getopt v0.0.0-20180729010549-6fdd0a2c7117/go.mod h1:85jBQOZwpVEaDAr341tbn15RS4fCAsIst0qp7i8ex1o=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=

swift/swift.go

@@ -0,0 +1,100 @@
+package swiftsource
+
+import (
+	"github.com/ncw/swift"
+	"github.com/xitongsys/parquet-go/source"
+)
+
+type SwiftFile struct {
+	Connection *swift.Connection
+
+	Container string
+	FilePath  string
+
+	FileReader *swift.ObjectOpenFile
+	FileWriter *swift.ObjectCreateFile
+}
+
+func newSwiftFile(containerName string, filePath string, conn *swift.Connection) *SwiftFile {
+	return &SwiftFile{
+		Connection: conn,
+		Container:  containerName,
+		FilePath:   filePath,
+	}
+}
+
+func NewSwiftFileReader(container string, filePath string, conn *swift.Connection) (source.ParquetFile, error) {
+	res := newSwiftFile(container, filePath, conn)
+	return res.Open(filePath)
+}
+
+func NewSwiftFileWriter(container string, filePath string, conn *swift.Connection) (source.ParquetFile, error) {
+	res := newSwiftFile(container, filePath, conn)
+	return res.Create(filePath)
+}
+
+func (file *SwiftFile) Open(name string) (source.ParquetFile, error) {
+	if name == "" {
+		name = file.FilePath
+	}
+
+	fr, _, err := file.Connection.ObjectOpen(file.Container, name, false, nil)
+	if err != nil {
+		return nil, err
+	}
+
+	res := &SwiftFile{
+		Connection: file.Connection,
+		Container:  file.Container,
+		FilePath:   name,
+		FileReader: fr,
+	}
+
+	return res, nil
+}
+
+func (file *SwiftFile) Create(name string) (source.ParquetFile, error) {
+	if name == "" {
+		name = file.FilePath
+	}
+
+	fw, err := file.Connection.ObjectCreate(file.Container, name, false, "", "", nil)
+	if err != nil {
+		return nil, err
+	}
+
+	res := &SwiftFile{
+		Connection: file.Connection,
+		Container:  file.Container,
+		FilePath:   name,
+		FileWriter: fw,
+	}
+
+	return res, nil
+}
+
+func (file *SwiftFile) Read(b []byte) (n int, err error) {
+	return file.FileReader.Read(b)
+}
+
+func (file *SwiftFile) Seek(offset int64, whence int) (int64, error) {
+	return file.FileReader.Seek(offset, whence)
+}
+
+func (file *SwiftFile) Write(p []byte) (n int, err error) {
+	return file.FileWriter.Write(p)
+}
+
+func (file *SwiftFile) Close() error {
+	if file.FileWriter != nil {
+		if err := file.FileWriter.Close(); err != nil {
+			return err
+		}
+	}
+	if file.FileReader != nil {
+		if err := file.FileReader.Close(); err != nil {
+			return err
+		}
+	}
+	return nil
+}