Hi Sagar, That's more a question for the ORC community, but AFAIK, the top-level type is always a struct because it needs to wrap the fields, e.g., struct(name:string, age:int)
Best, Fabian 2018-06-26 22:38 GMT+02:00 sagar loke <sagar...@gmail.com>: > @zhangminglei, > > Question about the schema for ORC format: > > 1. Does it always need to be of complex type "<Struct>" ? > > 2. Or can it be created with individual data types directly ? > eg. "name:string, age:int" ? > > > Thanks, > Sagar > > On Fri, Jun 22, 2018 at 11:56 PM, zhangminglei <18717838...@163.com> > wrote: > >> Yes, it should be exit. Thanks to Ted Yu. Very exactly! >> >> Cheers >> Zhangminglei >> >> 在 2018年6月23日，下午12:40，Ted Yu <yuzhih...@gmail.com> 写道： >> >> For #1, the word exist should be exit, right ? >> Thanks >> >> -------- Original message -------- >> From: zhangminglei <18717838...@163.com> >> Date: 6/23/18 10:12 AM (GMT+08:00) >> To: sagar loke <sagar...@gmail.com> >> Cc: dev <dev@flink.apache.org>, user <u...@flink.apache.org> >> Subject: Re: [Flink-9407] Question about proposed ORC Sink ! >> >> Hi, Sagar. >> >> 1. It solves the issue partially meaning files which have finished >> checkpointing don't show .pending status but the files which were in >> progress >> when the program exists are still in .pending state. >> >> >> Ans: >> >> Yea, Make the program exists and in that time if a checkpoint does not >> finished will lead the status keeps in .pending state then. Under the >> normal circumstances, the programs that running in the production env will >> never be stoped or existed if everything is fine. >> >> 2. Ideally, writer should work with default settings correct ? Meaning we >> don't have to explicitly set these parameters to make it work. >> Is this assumption correct ? >> >> >> Ans: >> >> Yes. Writer should work with default settings correct. >> Yes. We do not have to explicitly set these parameters to make it work. >> Yes. Assumption correct indeed. >> >> However, you know, flink is a real time streaming framework, so under >> normal circumstances,you don't really go to use the default settings when >> it comes to a specific business. Especially together work with *offline >> end*(Like hadoop mapreduce). In this case, you need to tell the offline >> end when time a bucket is close and when time the data for the specify >> bucket is ready. So, you can take a look on https://issues.apache.org/j >> ira/browse/FLINK-9609. >> >> Cheers >> Zhangminglei >> >> >> 在 2018年6月23日，上午8:23，sagar loke <sagar...@gmail.com> 写道： >> >> Hi Zhangminglei, >> >> Thanks for the reply. >> >> 1. It solves the issue partially meaning files which have finished >> checkpointing don't show .pending status but the files which were in >> progress >> when the program exists are still in .pending state. >> >> 2. Ideally, writer should work with default settings correct ? Meaning we >> don't have to explicitly set these parameters to make it work. >> Is this assumption correct ? >> >> Thanks, >> Sagar >> >> On Fri, Jun 22, 2018 at 3:25 AM, zhangminglei <18717838...@163.com> >> wrote: >> >>> Hi, Sagar. Please use the below code and you will find the part files >>> status from _part-0-107.in-progress to _part-0-107.pending and >>> finally to part-0-107. [For example], you need to run the program for a >>> while. However, we need set some parameters, like the following. Moreover, >>> *enableCheckpointing* IS also needed. I know why you always see the >>> *.pending* file since the below parameters default value is 60 seconds >>> even though you set the enableCheckpoint. So, that is why you can not see >>> the finished file status until 60 seconds passed. >>> >>> Attached is the ending on my end, and you will see what you want! >>> >>> Please let me know if you still have the problem. >>> >>> Cheers >>> Zhangminglei >>> >>> setInactiveBucketCheckInterval(2000) >>> .setInactiveBucketThreshold(2000); >>> >>> >>> public class TestOrc { >>> public static void main(String[] args) throws Exception { >>> StreamExecutionEnvironment env = >>> StreamExecutionEnvironment.getExecutionEnvironment(); >>> env.setParallelism(1); >>> env.enableCheckpointing(1000); >>> env.setStateBackend(new MemoryStateBackend()); >>> >>> String orcSchemaString = >>> "struct<name:string,age:int,married:boolean>"; >>> String path = "hdfs://10.199.196.0:9000/data/hive/man"; >>> >>> BucketingSink<Row> bucketingSink = new BucketingSink<>(path); >>> >>> bucketingSink >>> .setWriter(new OrcFileWriter<>(orcSchemaString)) >>> .setInactiveBucketCheckInterval(2000) >>> .setInactiveBucketThreshold(2000); >>> >>> DataStream<Row> dataStream = env.addSource(new ManGenerator()); >>> >>> dataStream.addSink(bucketingSink); >>> >>> env.execute(); >>> } >>> >>> public static class ManGenerator implements SourceFunction<Row> { >>> >>> @Override >>> public void run(SourceContext<Row> ctx) throws Exception { >>> for (int i = 0; i < 2147483000; i++) { >>> Row row = new Row(3); >>> row.setField(0, "Sagar"); >>> row.setField(1, 26 + i); >>> row.setField(2, false); >>> ctx.collect(row); >>> } >>> } >>> >>> @Override >>> public void cancel() { >>> >>> } >>> } >>> } >>> >>> <filestatus.jpg> >>> >>> >>> >>> 在 2018年6月22日，上午11:14，sagar loke <sagar...@gmail.com> 写道： >>> >>> Sure, we can solve it together :) >>> >>> Are you able to reproduce it ? >>> >>> Thanks, >>> Sagar >>> >>> On Thu, Jun 21, 2018 at 7:28 PM zhangminglei <18717838...@163.com> >>> wrote: >>> >>>> Sagar, flush will be called when do a checkpoint. Please see >>>> >>>> bucketState.currentFileValidLength = bucketState.writer.flush(); >>>> >>>> >>>> >>>> @Override >>>> public void snapshotState(FunctionSnapshotContext context) throws >>>> Exception { >>>> Preconditions.checkNotNull(restoredBucketStates, "The operator has not >>>> been properly initialized."); >>>> >>>> restoredBucketStates.clear(); >>>> >>>> synchronized (state.bucketStates) { >>>> int subtaskIdx = getRuntimeContext().getIndexOfThisSubtask(); >>>> >>>> for (Map.Entry<String, BucketState<T>> bucketStateEntry : >>>> state.bucketStates.entrySet()) { >>>> BucketState<T> bucketState = bucketStateEntry.getValue(); >>>> >>>> if (bucketState.isWriterOpen) { >>>> bucketState.currentFileValidLength = >>>> bucketState.writer.flush(); >>>> } >>>> >>>> synchronized (bucketState.pendingFilesPerCheckpoint) { >>>> >>>> bucketState.pendingFilesPerCheckpoint.put(context.getCheckpointId(), >>>> bucketState.pendingFiles); >>>> } >>>> bucketState.pendingFiles = new ArrayList<>(); >>>> } >>>> restoredBucketStates.add(state); >>>> >>>> if (LOG.isDebugEnabled()) { >>>> LOG.debug("{} idx {} checkpointed {}.", >>>> getClass().getSimpleName(), subtaskIdx, state); >>>> } >>>> } >>>> >>>> >>>> >>>> 在 2018年6月22日，上午10:21，sagar loke <sagar...@gmail.com> 写道： >>>> >>>> Thanks for replying. >>>> >>>> Yes, I tried with different values of checkpoint eg. 20, 100, 5000. >>>> >>>> env.enablecheckpointing(100); >>>> >>>> But in all the cases, I still see .pending state. >>>> >>>> Not sure if it’s related to flush() method from OrcFileWriter ? Which >>>> might not be getting called somehow ? >>>> >>>> Thanks, >>>> Sagar >>>> >>>> On Thu, Jun 21, 2018 at 7:17 PM zhangminglei <18717838...@163.com> >>>> wrote: >>>> >>>>> Hi，Sagar >>>>> >>>>> Please take a look at BucketingSink, It says that a file would keep >>>>> .pending status if you DO NOT do a checkpoint. Doc says, when a >>>>> checkpoint >>>>> is successful the currently pending file will be removed to {@code >>>>> finished}. >>>>> Take a try again. I think you should call the below method and see >>>>> what would happen on it. Anyway, I will also try that and see whether it >>>>> works. Please let me know if you still meet error. >>>>> >>>>> env.enableCheckpointing(200); >>>>> >>>>> /** >>>>> * The suffix for {@code pending} part files. These are closed files that >>>>> we are >>>>> * not currently writing to (inactive or reached {@link #batchSize}), but >>>>> which >>>>> * were not yet confirmed by a checkpoint. >>>>> */ >>>>> private static final String DEFAULT_PENDING_SUFFIX = ".pending"; >>>>> >>>>> <p>Part files can be in one of three states: {@code in-progress}, {@code >>>>> pending} or {@code finished}. >>>>> * The reason for this is how the sink works together with the >>>>> checkpointing mechanism to provide exactly-once >>>>> * semantics and fault-tolerance. The part file that is currently being >>>>> written to is {@code in-progress}. Once >>>>> * a part file is closed for writing it becomes {@code pending}. When a >>>>> checkpoint is successful the currently >>>>> * pending files will be moved to {@code finished}. >>>>> >>>>> >>>>> Cheers >>>>> Zhangminglei >>>>> >>>>> >>>>> >>>>> 在 2018年6月22日，上午4:46，sagar loke <sagar...@gmail.com> 写道： >>>>> >>>>> Thanks Zhangminglei for quick response. >>>>> >>>>> I tried the above code and I am seeing another issue where the files >>>>> created on hdfs are always in *.pending* state. >>>>> >>>>> Let me know if you can reproduce it ? >>>>> >>>>> Thanks, >>>>> Sagar >>>>> >>>>> On Thu, Jun 21, 2018 at 3:20 AM, zhangminglei <18717838...@163.com> >>>>> wrote: >>>>> >>>>>> Hi, Sagar >>>>>> >>>>>> I did a local test for that and it seems works fine for me. PR will >>>>>> be updated for [FLINK-9407] >>>>>> >>>>>> I will update the newest code to PR soon and below is the example I >>>>>> was using for my test. You can check it again. Hopes you can enjoy it! >>>>>> >>>>>> Cheers >>>>>> Zhangminglei. >>>>>> >>>>>> import org.apache.flink.streaming.api.datastream.DataStream; >>>>>> import >>>>>> org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; >>>>>> import org.apache.flink.streaming.api.functions.source.SourceFunction; >>>>>> import org.apache.flink.streaming.connectors.fs.bucketing.BucketingSink; >>>>>> import org.apache.flink.types.Row; >>>>>> >>>>>> public class TestOrc { >>>>>> public static void main(String[] args) throws Exception { >>>>>> StreamExecutionEnvironment env = >>>>>> StreamExecutionEnvironment.getExecutionEnvironment(); >>>>>> env.setParallelism(1); >>>>>> >>>>>> String orcSchemaString = >>>>>> "struct<name:string,age:int,married:boolean>"; >>>>>> String path = "hdfs://10.199.196.0:9000/data/hive/man"; >>>>>> >>>>>> BucketingSink<Row> bucketingSink = new BucketingSink<>(path); >>>>>> >>>>>> bucketingSink.setWriter(new OrcFileWriter<>(orcSchemaString)); >>>>>> >>>>>> DataStream<Row> dataStream = env.addSource(new ManGenerator()); >>>>>> >>>>>> dataStream.addSink(bucketingSink); >>>>>> >>>>>> env.execute(); >>>>>> } >>>>>> >>>>>> public static class ManGenerator implements SourceFunction<Row> { >>>>>> >>>>>> @Override >>>>>> public void run(SourceContext<Row> ctx) throws Exception { >>>>>> for (int i = 0; i < 3; i++) { >>>>>> Row row = new Row(3); >>>>>> row.setField(0, "Sagar"); >>>>>> row.setField(1, 26 + i); >>>>>> row.setField(2, false); >>>>>> ctx.collect(row); >>>>>> } >>>>>> } >>>>>> >>>>>> @Override >>>>>> public void cancel() { >>>>>> >>>>>> } >>>>>> } >>>>>> } >>>>>> >>>>>> >>>>>> >>>>>> 在 2018年6月21日，上午1:47，sagar loke <sagar...@gmail.com> 写道： >>>>>> >>>>>> Hi Zhangminglei, >>>>>> >>>>>> Question about https://issues.apache.org/jira/browse/FLINK-9407 >>>>>> >>>>>> I tried to use the code from PR and run it on local hdfs cluster to >>>>>> write some ORC data. >>>>>> >>>>>> But somehow this code is failing with following error: >>>>>> >>>>>> >>>>>> >>>>>> org.apache.hadoop.ipc.RemoteException(org.apache.hadoop.hdfs >>>>>>> .protocol.AlreadyBeingCreatedException): Failed to CREATE_FILE >>>>>>> /tmp/hivedatanew2/2018-06-20/10/34/_part-0-0.in-progress for >>>>>>> DFSClient_NONMAPREDUCE_73219864_36 on 127.0.0.1 because this file >>>>>>> lease is currently owned by DFSClient_NONMAPREDUCE_-1374584007_36 >>>>>>> on 127.0.0.1 >>>>>> >>>>>> at org.apache.hadoop.hdfs.server.namenode.FSNamesystem.recoverL >>>>>>> easeInternal(FSNamesystem.java:2500) >>>>>> >>>>>> >>>>>> I understand that this error is related to Hadoop but somehow I get >>>>>> this error only when executing the code from this PR. >>>>>> >>>>>> I had created very crude way to write ORC file to HDFS as per >>>>>> follows. Below code works alright and does not throw above error. >>>>>> >>>>>> import org.apache.flink.streaming.connectors.fs.Writer; >>>>>>> import org.apache.hadoop.fs.FileSystem; >>>>>>> import org.apache.hadoop.fs.Path; >>>>>>> import org.apache.hadoop.hive.ql.exec.vector.LongColumnVector; >>>>>>> import org.apache.hadoop.hive.ql.exec.vector.VectorizedRowBatch; >>>>>>> import org.apache.orc.OrcFile; >>>>>>> import org.apache.orc.TypeDescription; >>>>>>> import org.apache.hadoop.conf.Configuration; >>>>>>> >>>>>>> import java.io.IOException; >>>>>>> >>>>>>> public class FlinkOrcWriterV1<T> implements >>>>>>> org.apache.flink.streaming.connectors.fs.Writer<T> { >>>>>>> >>>>>>> private transient org.apache.orc.Writer orcWriter; >>>>>>> String schema; >>>>>>> TypeDescription typeDescriptionschema;//"struct<x:int,y:int>" >>>>>>> String basePath; >>>>>>> >>>>>>> public FlinkOrcWriterV1(String schema) { >>>>>>> this.schema = schema; >>>>>>> this.typeDescriptionschema = TypeDescription.fromString(schema); >>>>>>> } >>>>>>> >>>>>>> @Override >>>>>>> public void open(FileSystem fs, Path path) throws IOException { >>>>>>> Configuration conf = new Configuration(); >>>>>>> orcWriter = OrcFile.createWriter(new >>>>>>> Path("hdfs://localhost:9000/tmp/hivedata3/"), >>>>>>> OrcFile.writerOptions(conf) >>>>>>> .setSchema(typeDescriptionschema)); >>>>>>> } >>>>>>> >>>>>>> @Override >>>>>>> public long flush() throws IOException { >>>>>>> return orcWriter.writeIntermediateFooter(); >>>>>>> } >>>>>>> >>>>>>> @Override >>>>>>> public long getPos() throws IOException { >>>>>>> return orcWriter.getRawDataSize(); >>>>>>> } >>>>>>> >>>>>>> @Override >>>>>>> public void close() throws IOException { >>>>>>> orcWriter.close(); >>>>>>> } >>>>>>> >>>>>>> @Override >>>>>>> public void write(T element) throws IOException { >>>>>>> VectorizedRowBatch batch = >>>>>>> typeDescriptionschema.createRowBatch(10); >>>>>>> LongColumnVector x = (LongColumnVector) batch.cols[0]; >>>>>>> LongColumnVector y = (LongColumnVector) batch.cols[1]; >>>>>>> for(int r=0; r < 10; ++r) { >>>>>>> int row = batch.size++; >>>>>>> x.vector[row] = r; >>>>>>> y.vector[row] = r * 3; >>>>>>> // If the batch is full, write it out and start over. >>>>>>> if (batch.size == batch.getMaxSize()) { >>>>>>> orcWriter.addRowBatch(batch); >>>>>>> batch.reset(); >>>>>>> } >>>>>>> } >>>>>>> if (batch.size != 0) { >>>>>>> orcWriter.addRowBatch(batch); >>>>>>> batch.reset(); >>>>>>> } >>>>>>> } >>>>>>> >>>>>>> @Override >>>>>>> public FlinkOrcWriterV1<T> duplicate() { >>>>>>> return new FlinkOrcWriterV1<>(schema); >>>>>>> } >>>>>>> } >>>>>>> >>>>>>> >>>>>>> >>>>>> >>>>>> Not sure, if the error is related to any of the hadoop dependencies >>>>>> or something else ? >>>>>> >>>>>> Can you please look into it and let me know if you can reproduce it >>>>>> on your end too ? >>>>>> >>>>>> By the way, following are my dependencies in my project: >>>>>> >>>>>> dependencies { >>>>>>> >>>>>>> compile 'org.apache.flink:flink-java:1.4.2' >>>>>>> >>>>>>> compile 'org.apache.flink:flink-runtime_2.11:1.4.2' >>>>>>> >>>>>>> compile 'org.apache.flink:flink-streaming-java_2.11:1.4.2' >>>>>>> >>>>>>> compile 'org.apache.flink:flink-connector-kafka-0.11_2.11:1.4.2' >>>>>>> >>>>>>> compile 'org.apache.flink:flink-connec >>>>>>>> tor-elasticsearch5_2.11:1.4.2' >>>>>>> >>>>>>> compile 'io.confluent:kafka-avro-serializer:3.3.0' >>>>>>> >>>>>>> compile 'org.apache.flink:flink-avro:1.4.2' >>>>>>> >>>>>>> compile group: 'org.apache.kafka', name: 'kafka_2.11', version: >>>>>>>> '1.1.0' >>>>>>> >>>>>>> compile group: 'org.apache.flink', name: 'flink-shaded-hadoop2', >>>>>>>> version: '1.4.2' >>>>>>> >>>>>>> compile 'org.apache.flink:flink-connector-filesystem_2.11:1.4.2' >>>>>>> >>>>>>> compile group: 'org.apache.flink', name: 'flink-jdbc', version: >>>>>>>> '1.4.2' >>>>>>> >>>>>>> compile group: 'org.apache.flink', name: 'flink-table_2.11', >>>>>>>> version: '1.4.2' >>>>>>> >>>>>>> compile group: 'org.apache.orc', name: 'orc-core', version: >>>>>>>> '1.5.1' >>>>>>> >>>>>>> compile group: 'org.apache.parquet', name: 'parquet-avro', >>>>>>>> version: '1.10.0' >>>>>>> >>>>>>> compile group: 'org.apache.parquet', name: 'parquet-common', >>>>>>>> version: '1.10.0' >>>>>>> >>>>>>> compile group: 'org.apache.flink', name: 'flink-orc_2.11', >>>>>>>> version: '1.4.2' >>>>>>> >>>>>>> testCompile group: 'junit', name: 'junit', version: '4.12' >>>>>>> >>>>>>> } >>>>>>> >>>>>>> >>>>>> >>>>>> -- >>>>>> Thanks, >>>>>> Sagar. >>>>>> >>>>>> >>>>>> >>>>> >>>>> >>>>> -- >>>>> Regards, >>>>> SAGAR. >>>>> >>>>> >>>>> -- >>>> Cheers, >>>> Sagar >>>> >>>> >>>> -- >>> Cheers, >>> Sagar >>> >>> >>> >> >> >> -- >> Regards, >> SAGAR. >> >> >> >> > > > -- > Regards, > SAGAR. >
