Skip to content

Commit

Permalink
Merge pull request #170 from DanteNiewenhuis/export-format
Browse files Browse the repository at this point in the history 
updaded output format of parquet exporter
  • Loading branch information
@DanteNiewenhuis
DanteNiewenhuis authored Nov 21, 2023
2 parents 05141e9 + ed06c38 commit 05215c1
Show file tree
Hide file tree
Showing 2 changed files with 43 additions and 38 deletions.
9 changes: 4 additions & 5 deletions ...te/src/main/kotlin/org/opendc/experiments/compute/export/parquet/ParquetHostDataWriter.kt
View file
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ package org.opendc.experiments.compute.export.parquet
import org.apache.hadoop.conf.Configuration
import org.apache.parquet.hadoop.ParquetWriter
import org.apache.parquet.hadoop.api.WriteSupport
import org.apache.parquet.io.api.Binary
import org.apache.parquet.io.api.RecordConsumer
import org.apache.parquet.schema.LogicalTypeAnnotation
import org.apache.parquet.schema.MessageType
Expand All @@ -33,7 +34,6 @@ import org.apache.parquet.schema.Types
import org.opendc.experiments.compute.telemetry.table.HostTableReader
import org.opendc.trace.util.parquet.LocalParquetWriter
import java.io.File
import java.util.UUID

/**
* A Parquet event writer for [HostTableReader]s.
Expand Down Expand Up @@ -75,7 +75,7 @@ public class ParquetHostDataWriter(path: File, bufferSize: Int) :
consumer.endField("timestamp", 0)

consumer.startField("host_id", 1)
consumer.addBinary(UUID.fromString(data.host.id).toBinary())
consumer.addBinary(Binary.fromString(data.host.id))
consumer.endField("host_id", 1)

consumer.startField("cpu_count", 2)
Expand Down Expand Up @@ -169,9 +169,8 @@ public class ParquetHostDataWriter(path: File, bufferSize: Int) :
.`as`(LogicalTypeAnnotation.timestampType(true, LogicalTypeAnnotation.TimeUnit.MILLIS))
.named("timestamp"),
Types
.required(PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY)
.length(16)
.`as`(LogicalTypeAnnotation.uuidType())
.required(PrimitiveType.PrimitiveTypeName.BINARY)
.`as`(LogicalTypeAnnotation.stringType())
.named("host_id"),
Types
.required(PrimitiveType.PrimitiveTypeName.INT32)
Expand Down
72 changes: 39 additions & 33 deletions .../src/main/kotlin/org/opendc/experiments/compute/export/parquet/ParquetServerDataWriter.kt
View file
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ package org.opendc.experiments.compute.export.parquet
import org.apache.hadoop.conf.Configuration
import org.apache.parquet.hadoop.ParquetWriter
import org.apache.parquet.hadoop.api.WriteSupport
import org.apache.parquet.io.api.Binary
import org.apache.parquet.io.api.RecordConsumer
import org.apache.parquet.schema.LogicalTypeAnnotation
import org.apache.parquet.schema.MessageType
Expand All @@ -33,7 +34,6 @@ import org.apache.parquet.schema.Types
import org.opendc.experiments.compute.telemetry.table.ServerTableReader
import org.opendc.trace.util.parquet.LocalParquetWriter
import java.io.File
import java.util.UUID

/**
* A Parquet event writer for [ServerTableReader]s.
Expand Down Expand Up @@ -76,64 +76,68 @@ public class ParquetServerDataWriter(path: File, bufferSize: Int) :
consumer.endField("timestamp", 0)

consumer.startField("server_id", 1)
consumer.addBinary(UUID.fromString(data.server.id).toBinary())
consumer.addBinary(Binary.fromString(data.server.id))
consumer.endField("server_id", 1)

consumer.startField("server_name", 2)
consumer.addBinary(Binary.fromString(data.server.name))
consumer.endField("server_name", 2)

val hostId = data.host?.id
if (hostId != null) {
consumer.startField("host_id", 2)
consumer.addBinary(UUID.fromString(hostId).toBinary())
consumer.endField("host_id", 2)
consumer.startField("host_id", 3)
consumer.addBinary(Binary.fromString(hostId))
consumer.endField("host_id", 3)
}

consumer.startField("mem_capacity", 3)
consumer.startField("mem_capacity", 4)
consumer.addLong(data.server.memCapacity)
consumer.endField("mem_capacity", 3)
consumer.endField("mem_capacity", 4)

consumer.startField("cpu_count", 4)
consumer.startField("cpu_count", 5)
consumer.addInteger(data.server.cpuCount)
consumer.endField("cpu_count", 4)
consumer.endField("cpu_count", 5)

consumer.startField("cpu_limit", 5)
consumer.startField("cpu_limit", 6)
consumer.addDouble(data.cpuLimit)
consumer.endField("cpu_limit", 5)
consumer.endField("cpu_limit", 6)

consumer.startField("cpu_time_active", 6)
consumer.startField("cpu_time_active", 7)
consumer.addLong(data.cpuActiveTime)
consumer.endField("cpu_time_active", 6)
consumer.endField("cpu_time_active", 7)

consumer.startField("cpu_time_idle", 7)
consumer.startField("cpu_time_idle", 8)
consumer.addLong(data.cpuIdleTime)
consumer.endField("cpu_time_idle", 7)
consumer.endField("cpu_time_idle", 8)

consumer.startField("cpu_time_steal", 8)
consumer.startField("cpu_time_steal", 9)
consumer.addLong(data.cpuStealTime)
consumer.endField("cpu_time_steal", 8)
consumer.endField("cpu_time_steal", 9)

consumer.startField("cpu_time_lost", 9)
consumer.startField("cpu_time_lost", 10)
consumer.addLong(data.cpuLostTime)
consumer.endField("cpu_time_lost", 9)
consumer.endField("cpu_time_lost", 10)

consumer.startField("uptime", 10)
consumer.startField("uptime", 11)
consumer.addLong(data.uptime)
consumer.endField("uptime", 10)
consumer.endField("uptime", 11)

consumer.startField("downtime", 11)
consumer.startField("downtime", 12)
consumer.addLong(data.downtime)
consumer.endField("downtime", 11)
consumer.endField("downtime", 12)

val provisionTime = data.provisionTime
if (provisionTime != null) {
consumer.startField("provision_time", 12)
consumer.startField("provision_time", 13)
consumer.addLong(provisionTime.toEpochMilli())
consumer.endField("provision_time", 12)
consumer.endField("provision_time", 13)
}

val bootTime = data.bootTime
if (bootTime != null) {
consumer.startField("boot_time", 13)
consumer.startField("boot_time", 14)
consumer.addLong(bootTime.toEpochMilli())
consumer.endField("boot_time", 13)
consumer.endField("boot_time", 14)
}

consumer.endMessage()
Expand All @@ -151,14 +155,16 @@ public class ParquetServerDataWriter(path: File, bufferSize: Int) :
.`as`(LogicalTypeAnnotation.timestampType(true, LogicalTypeAnnotation.TimeUnit.MILLIS))
.named("timestamp"),
Types
.required(PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY)
.length(16)
.`as`(LogicalTypeAnnotation.uuidType())
.required(PrimitiveType.PrimitiveTypeName.BINARY)
.`as`(LogicalTypeAnnotation.stringType())
.named("server_id"),
Types
.optional(PrimitiveType.PrimitiveTypeName.FIXED_LEN_BYTE_ARRAY)
.length(16)
.`as`(LogicalTypeAnnotation.uuidType())
.required(PrimitiveType.PrimitiveTypeName.BINARY)
.`as`(LogicalTypeAnnotation.stringType())
.named("server_name"),
Types
.optional(PrimitiveType.PrimitiveTypeName.BINARY)
.`as`(LogicalTypeAnnotation.stringType())
.named("host_id"),
Types
.required(PrimitiveType.PrimitiveTypeName.INT64)
Expand Down

0 comments on commit 05215c1

Please sign in to comment.