Production Use about parquet-go HOT 33 CLOSED

package main

import (
	"encoding/csv"
	"fmt"
	"github.com/satori/go.uuid"
	. "github.com/xitongsys/parquet-go/ParquetFile"
	. "github.com/xitongsys/parquet-go/Plugin/CSVWriter"
	"log"
	"os"
	"strings"
	"sync"
)

var (
	workers       = make(map[string]chan []string)
	parquetWriter = make(map[string]*CSVWriter)
	csvWriter     = make(map[string]*csv.Writer)
	wg            = sync.WaitGroup{}
	mu            = sync.Mutex{}
	localPath     = "./"
	path          = localPath
	md            = []MetadataType{
		{Type: "UTF8", Name: "Name"},
		{Type: "INT32", Name: "Age"},
		{Type: "INT64", Name: "Id"},
		{Type: "FLOAT", Name: "Weight"},
		//{Type: "BOOLEAN", Name: "Sex"},
	}
)

func writerT(csvRaw string, tableMetaData []MetadataType) {
	defer wg.Wait()
	tableMetaData = tableMetaData
	s := strings.NewReader(csvRaw)
	r := csv.NewReader(s)

	lines, err := r.ReadAll()
	if err != nil {
		log.Fatalf("error reading all lines: %v", err)
	}
	numLines := len(lines) - 1

	for idx, rec := range lines {
		if idx == numLines {
			stopWorkers()
			return
		}
		process(rec)
	}
}

func process(rec []string) {
	l := len(rec)
	part := rec[l-1]

	if c, ok := workers[part]; ok {
		c <- rec
	} else {
		nc := make(chan []string)
		workers[part] = nc

		u := fmt.Sprintf("%v", uuid.NewV4())
		csvName := localPath + u + ".csv"
		csvFile, _ := os.OpenFile(csvName, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
		csvWriter[part] = csv.NewWriter(csvFile)

		fName := localPath + u + ".parq"
		pf, _ := NewLocalFileWriter(fName)
		parquetWriter[part], _ = NewCSVWriter(md, pf, 10)

		go workerP(nc)
		nc <- rec
	}
}

func workerP(c chan []string) {
	wg.Add(1)
	defer wg.Done()

	var pw *CSVWriter
	var writer *csv.Writer

	part := [][]string{}
	for {
		// wait for a rec or close(chan)
		rec, ok := <-c
		if ok {
			key := rec[len(rec)-1]
			pw = parquetWriter[key]
			writer = csvWriter[key]

			mu.Lock()
			// PARQUET
			row := make([]*string, len(rec))
			for j := 0; j < len(rec); j++ {
				row[j] = &rec[j]
			}
			// drop partition col
			row = row[:len(row)-1]
			pw.WriteString(row)

			// CSV
			if err := writer.Write(rec); err != nil {
				log.Fatal(err)
			}
			mu.Unlock()
			part = append(part, rec)

		} else {
			if len(part) <= 0 {
				break
			}

			key := part[0][len(part[0])-1]
			pw = parquetWriter[key]
			writer = csvWriter[key]

			mu.Lock()
			// PARQUET
			for _, p := range part {
				row := make([]*string, len(p))
				for j := 0; j < len(p); j++ {
					row[j] = &p[j]
				}
				pw.WriteString(row[:len(row)-1])
			}

			// CSV
			for _, p := range part {
				if err := writer.Write(p); err != nil {
					log.Fatal(err)
				}
				writer.Flush()
			}
			mu.Unlock()

			break
		}
	}
	pw.Flush(true)
	pw.WriteStop()
}

// simply signals to workers to stop
func stopWorkers() {
	for _, c := range workers {
		// signal to all workers to exit
		close(c)
	}
}

func main() {
	num := 10
	res := ""
	for i := 0; i < num; i++ {
		res += fmt.Sprintf("%s_%d", "Student Name", i) + "," +
			fmt.Sprintf("%d", 20+i%5) + "," +
			fmt.Sprintf("%d", i) + "," +
			fmt.Sprintf("%f", 50.0+float32(i)*0.1) + "," +
			fmt.Sprintf("%t", i%2 == 0) + "\n"
	}
	log.Println(res)
	writerT(res, md)
}

rob,pike,50
jane,doe,33
foo,bar,22

CREATE EXTERNAL TABLE temp.parqtest (`First` STRING, `Last` STRING, `Age` INT) 
STORED AS PARQUET 
LOCATION  's3a://path/to/parquet/directory/'
;

$ parquet-tools schema test.parquet
message parquet_go_root {
  required int64 Person_age;
  required binary Last_name (UTF8);
  required binary First_name (UTF8);
}

$ parquet-tools head -n 5 test.parquet
Person_age = 50
Last_name = pike
First_name = rob
... etc

CREATE EXTERNAL TABLE staging.ptest 
(Person_age INT, Last_name STRING, First_name STRING)
--( First_name STRING, Last_name STRING, Person_age INT)
STORED AS PARQUET
LOCATION  's3a://<my_bucket>/path/to/test_parq/'
;
REFRESH staging.ptest;

File 's3a://<my_bucket>/path/to/test_parq/test.parquet' uses an unsupported encoding: DELTA_BINARY_PACKED for column 'First'.

$ parquet-tools meta test.parquet
creator:     null

file schema: parquet_go_root
---------------------------------------------------------------------------------------------------------------------------------------
Person_age:  REQUIRED INT64 R:0 D:0
Last_name:   REQUIRED BINARY O:UTF8 R:0 D:0
First_name:  REQUIRED BINARY O:UTF8 R:0 D:0

row group 1: RC:3 TS:438
---------------------------------------------------------------------------------------------------------------------------------------
First_name:   BINARY SNAPPY DO:0 FPO:0 SZ:150/144/0.96 VC:3 ENC:PLAIN,DELTA_BINARY_PACKED,RLE,BIT_PACKED
Person_age:   INT64 SNAPPY DO:0 FPO:150 SZ:147/141/0.96 VC:3 ENC:PLAIN,DELTA_BINARY_PACKED,RLE,BIT_PACKED
Last_name:    BINARY SNAPPY DO:0 FPO:297 SZ:141/135/0.96 VC:3 ENC:PLAIN,DELTA_BINARY_PACKED,RLE,BIT_PACKED

package main
import (
	"github.com/xitongsys/parquet-go/parquet"
)
type Record struct {
	Cols []Column
}
type Column struct {
	Value        string
	PqType     parquet.ConvertedType
}

func (self *PqFile) CreateOrAppend(name string) (ParquetFile, error) {
	//file, err := os.Create(name)
	file, err := os.OpenFile(name, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
	myFile := new(PqFile)
	myFile.File = file
	return myFile, err
}

func (self *PqFile) Create(name string) (ParquetFile, error) {
	//file, err := os.Create(name)
	file, err := os.OpenFile(name, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
	myFile := new(PqFile)
	myFile.File = file
	return myFile, err
}

package main

import (
	"os"
	"fmt"
	"log"
	"sync"
	"strings"
	"crypto/rand"
	"encoding/csv"
	. "github.com/xitongsys/parquet-go/ParquetHandler"
	. "github.com/xitongsys/parquet-go/Plugin/CSVWriter"
)

var (
	// list of channels to communicate with workers
	// workers accessed synchronousely no mutex required
	workers = make(map[string]chan []string)

	// wg is to make sure all workers done before exiting main
	wg = sync.WaitGroup{}

	// mu used only for sequential printing, not relevant for program logic
	mu = sync.Mutex{}
	path = localPath
)

func writerT(csvRaw string, tableMetaData []MetadataType) {

	// wait for all workers to finish up before exit
	defer wg.Wait()

	// Set local instance of tableMetaData from what was passed in
	tableMetaData = tableMetaData
	s := strings.NewReader(csvRaw)
	r := csv.NewReader(s)

	lines, err := r.ReadAll()
	if err != nil {
		log.Fatalf("error reading all lines: %v", err)
	}
	numLines := len(lines) -1

	for idx, rec := range lines {
		if idx == numLines {
			stopWorkers()
			return
		}
		process(rec)
	}
}

func process(rec []string) {
	l := len(rec)
	part := rec[l-1]

	if c, ok := workers[part]; ok {
		// send rec to worker
		c <- rec
	} else {
		// if no worker for the partition

		// make a chan
		nc := make(chan []string)
		workers[part] = nc

		// start worker with this chan
		go workerP(nc)

		// send rec to worker via chan
		nc <- rec
	}
}

func workerP(c chan []string) {

	// wg.Done signals to main worker completion
	wg.Add(1)
	defer wg.Done()

	u, _ := newUUID()

	// CSV
	csvName := localPath + u + ".csv"
	// If the file doesn't exist, create it, or append to the file
	csvFile, err := os.OpenFile(csvName, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
	if err != nil {
		log.Fatal(err)
	}
	writer := csv.NewWriter(csvFile)

	fName := localPath + u + ".parq"
	fmt.Println(fName)

	// PARQUET
	var f ParquetFile
	f = &PqFile{}
	f, err = f.Create(fName)
	if err != nil {
		fmt.Println(err)
	}
	ph := NewCSVWriterHandler()
	ph.WriteInit(tableMetaData, f, 10, 30)

	part := [][]string{}
	for {
		// wait for a rec or close(chan)
		rec, ok := <-c
		if ok {
			// save the rec
			// instead of accumulation in memory
			// this can be saved to file directly

			mu.Lock()
			row := make([]*string, len(rec))
			for j := 0; j < len(rec); j++ {
				row[j] = &rec[j]
			}
			ph.Write(row)

			// CSV
			if err := writer.Write(rec); err != nil {
				log.Fatal(err)
			}
			mu.Unlock()

			part = append(part, rec)
		} else {
			// channel closed on EOF
			// dump partition
			// locks ensures sequential printing
			// not a required for independent files

			mu.Lock()
			row := make([]*string, len(rec))
			for j := 0; j < len(rec); j++ {
				row[j] = &rec[j]
			}
			ph.Write(row)

			// CSV
			for _, p := range part {
				if err := writer.Write(p); err != nil {
					log.Fatal(err)
				}
				writer.Flush()
			}
			mu.Unlock()

			return
		}
	}
	ph.Flush()
	ph.WriteStop()
	f.Close()
	csvFile.Close()
}

// simply signals to workers to stop
func stopWorkers() {
	for _, c := range workers {
		// signal to all workers to exit
		close(c)
	}
}

// newUUID generates a random UUID according to RFC 4122
func newUUID() (string, error) {
	uuid := make([]byte, 16)
	n, err := io.ReadFull(rand.Reader, uuid)
	if n != len(uuid) || err != nil {
		return "", err
	}
	// variant bits; see section 4.1.1
	uuid[8] = uuid[8]&^0xc0 | 0x80
	// version 4 (pseudo-random); see section 4.1.3
	uuid[6] = uuid[6]&^0xf0 | 0x40
	return fmt.Sprintf("%x-%x-%x-%x-%x", uuid[0:4], uuid[4:6], uuid[6:8], uuid[8:10], uuid[10:]), nil
}

row := make([]*string, len(rec))   
for j := 0; j < len(rec); j++ {   
       row[j] = &rec[j]   
}   
ph.Write(row)  

package main

import (
	"os"
	//"io"
	"fmt"
	"log"
	"sync"
	//"reflect"
	"strings"
	"encoding/csv"
	. "github.com/xitongsys/parquet-go/ParquetHandler"
	. "github.com/xitongsys/parquet-go/Plugin/CSVWriter"
)

var (
	// list of channels to communicate with workers
	// workers accessed synchronousely no mutex required
	workers = make(map[string]chan []string)

	// wg is to make sure all workers done before exiting main
	wg = sync.WaitGroup{}

	// mu used only for sequential printing, not relevant for program logic
	mu = sync.Mutex{}
	path = localPath

	// Moved to global scope
	//tableMetaData []MetadataType
)

func writerT(csvRaw string, tableMetaData []MetadataType) {

	// wait for all workers to finish up before exit
	defer wg.Wait()

	// Set local instance of tableMetaData from what was passed in
	tableMetaData = tableMetaData
	s := strings.NewReader(csvRaw)
	r := csv.NewReader(s)

	lines, err := r.ReadAll()
	if err != nil {
		log.Fatalf("error reading all lines: %v", err)
	}
	numLines := len(lines) -1

	for idx, rec := range lines {
		if idx == numLines {
			stopWorkers()
			return
		}
		process(rec)
	}
}

func process(rec []string) {
	l := len(rec)
	part := rec[l-1]

	if c, ok := workers[part]; ok {
		// send rec to worker
		c <- rec
	} else {
		// if no worker for the partition

		// make a chan
		nc := make(chan []string)
		workers[part] = nc

		// start worker with this chan
		go workerP(nc)

		// send rec to worker via chan
		nc <- rec
	}
}

func workerP(c chan []string) {

	// wg.Done signals to main worker completion
	wg.Add(1)
	defer wg.Done()

	u, _ := newUUID()

	// CSV
	csvName := localPath + u + ".csv"
	// If the file doesn't exist, create it, or append to the file
	csvFile, err := os.OpenFile(csvName, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0644)
	if err != nil {
		log.Fatal(err)
	}
	writer := csv.NewWriter(csvFile)

	fName := localPath + u + ".parq"
	fmt.Println(fName)

	// PARQUET
	var pf ParquetFile
	//var err error
	pf = &MyFile{}
	pf, err = pf.Create(fName)
	if err != nil {
		fmt.Println(err)
	}

    ph := NewCSVWriterHandler()
	fmt.Println("TMD:", tableMetaData)
    ph.WriteInit(tableMetaData, pf, 10, 30)

	part := [][]string{}
	for {
		// wait for a rec or close(chan)
		rec, ok := <-c
		if ok {
			// save the rec
			// instead of accumulation in memory
			// this can be saved to file directly

			mu.Lock()
			// PARQUET
			row := make([]*string, len(rec))
			for j := 0; j < len(rec); j++ {
				row[j] = &rec[j]
			}

			// drop partition col
			row = row[:len(row) - 1]
			ph.WriteString(row)

			// cannot use rec (type []string) as type []*string in argument to ph.WriteString
			//rec = rec[:len(rec) - 1]
			//ph.WriteString(rec)

			// CSV
			if err := writer.Write(rec); err != nil {
				log.Fatal(err)
			}
			mu.Unlock()

			part = append(part, rec)

		} else {
			// channel closed on EOF
			// dump partition
			// locks ensures sequential printing
			// not a required for independent files

			mu.Lock()
			// PARQUET
			for _, p := range part {
				row := make([]*string, len(p))
				for j := 0; j < len(p); j++ {
					row[j] = &p[j]
					// drop partition col
					noPart := row[:len(row) - 1]
					ph.WriteString(noPart)
				}
			}

			// CSV
			for _, p := range part {
				if err := writer.Write(p); err != nil {
					log.Fatal(err)
				}
				writer.Flush()
			}
			mu.Unlock()

			return
		}
	}
	ph.Flush()
	ph.WriteStop()
	pf.Close()
	csvFile.Close()
}

// simply signals to workers to stop
func stopWorkers() {
	for _, c := range workers {
		// signal to all workers to exit
		close(c)
	}
}

$ parquet-tools head -n 10 myfile.parq
ingest_datetime = 2017
event_datetime = 2017
... etc, more columns

$ parquet-tools schema myfile.parq
 optional int32 ingest_datetime (DATE);
 optional int32 event_datetime (DATE);
... etc, more columns

+-----------+---+---+------+-----+----------+
|       Name|Age| Id|Weight|  Sex|       Day|
+-----------+---+---+------+-----+----------+
|StudentName| 20|  0|  50.0| true|2017-11-28|
|StudentName| 21|  1|  50.1|false|2017-11-28|
|StudentName| 22|  2|  50.2| true|2017-11-28|
|StudentName| 23|  3|  50.3|false|2017-11-28|
|StudentName| 24|  4|  50.4| true|2017-11-28|
|StudentName| 20|  5|  50.5|false|2017-11-28|
|StudentName| 21|  6|  50.6| true|2017-11-28|
|StudentName| 22|  7|  50.7|false|2017-11-28|
|StudentName| 23|  8|  50.8| true|2017-11-28|
|StudentName| 24|  9|  50.9|false|2017-11-28|
+-----------+---+---+------+-----+----------+