001 /*
002 * Copyright (c) 2007-2014 Concurrent, Inc. All Rights Reserved.
003 *
004 * Project and contact information: http://www.cascading.org/
005 *
006 * This file is part of the Cascading project.
007 *
008 * Licensed under the Apache License, Version 2.0 (the "License");
009 * you may not use this file except in compliance with the License.
010 * You may obtain a copy of the License at
011 *
012 * http://www.apache.org/licenses/LICENSE-2.0
013 *
014 * Unless required by applicable law or agreed to in writing, software
015 * distributed under the License is distributed on an "AS IS" BASIS,
016 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
017 * See the License for the specific language governing permissions and
018 * limitations under the License.
019 */
020
021 package cascading.tap;
022
023 import java.io.IOException;
024 import java.io.Serializable;
025 import java.util.Set;
026
027 import cascading.flow.Flow;
028 import cascading.flow.FlowElement;
029 import cascading.flow.FlowException;
030 import cascading.flow.FlowProcess;
031 import cascading.flow.planner.Scope;
032 import cascading.pipe.Pipe;
033 import cascading.property.ConfigDef;
034 import cascading.scheme.Scheme;
035 import cascading.tuple.Fields;
036 import cascading.tuple.FieldsResolverException;
037 import cascading.tuple.Tuple;
038 import cascading.tuple.TupleEntryCollector;
039 import cascading.tuple.TupleEntryIterator;
040 import cascading.util.Util;
041
042 /**
043 * A Tap represents the physical data source or sink in a connected {@link cascading.flow.Flow}.
044 * </p>
045 * That is, a source Tap is the head end of a connected {@link Pipe} and {@link Tuple} stream, and
046 * a sink Tap is the tail end. Kinds of Tap types are used to manage files from a local disk,
047 * distributed disk, remote storage like Amazon S3, or via FTP. It simply abstracts
048 * out the complexity of connecting to these types of data sources.
049 * <p/>
050 * A Tap takes a {@link Scheme} instance, which is used to identify the type of resource (text file, binary file, etc).
051 * A Tap is responsible for how the resource is reached.
052 * <p/>
053 * By default when planning a Flow, Tap equality is a function of the {@link #getIdentifier()} and {@link #getScheme()}
054 * values. That is, two Tap instances are the same Tap instance if they sink/source the same resource and sink/source
055 * the same fields.
056 * <p/>
057 * Some more advanced taps, like a database tap, may need to extend equality to include any filtering, like the
058 * {@code where} clause in a SQL statement so two taps reading from the same SQL table aren't considered equal.
059 * <p/>
060 * Taps are also used to determine dependencies between two or more {@link Flow} instances when used with a
061 * {@link cascading.cascade.Cascade}. In that case the {@link #getFullIdentifier(Object)} value is used and the Scheme
062 * is ignored.
063 */
064 public abstract class Tap<Config, Input, Output> implements FlowElement, Serializable
065 {
066 /** Field scheme */
067 private Scheme<Config, Input, Output, ?, ?> scheme;
068
069 /** Field mode */
070 SinkMode sinkMode = SinkMode.KEEP;
071
072 private ConfigDef configDef;
073
074 private ConfigDef processConfigDef;
075
076 /** Field id */
077 private String id = null;
078 /** Field trace */
079 private final String trace = Util.captureDebugTrace( getClass() );
080
081 /**
082 * Convenience function to make an array of Tap instances.
083 *
084 * @param taps of type Tap
085 * @return Tap array
086 */
087 public static Tap[] taps( Tap... taps )
088 {
089 return taps;
090 }
091
092 /**
093 * Creates and returns a unique ID for the given Tap, this value is cached and may be used to uniquely identify
094 * the Tap instance in properties files etc.
095 * <p/>
096 * This value is generally reproducible assuming the Tap identifier and the Scheme source and sink Fields remain consistent.
097 *
098 * @param tap of type Tap
099 * @return of type String
100 */
101 public static synchronized String id( Tap tap )
102 {
103 if( tap.id == null )
104 tap.id = Util.createUniqueID();
105
106 return tap.id;
107 }
108
109 protected Tap()
110 {
111 }
112
113 protected Tap( Scheme<Config, Input, Output, ?, ?> scheme )
114 {
115 this.setScheme( scheme );
116 }
117
118 protected Tap( Scheme<Config, Input, Output, ?, ?> scheme, SinkMode sinkMode )
119 {
120 this.setScheme( scheme );
121 this.sinkMode = sinkMode;
122 }
123
124 protected void setScheme( Scheme<Config, Input, Output, ?, ?> scheme )
125 {
126 this.scheme = scheme;
127 }
128
129 /**
130 * Method getScheme returns the scheme of this Tap object.
131 *
132 * @return the scheme (type Scheme) of this Tap object.
133 */
134 public Scheme<Config, Input, Output, ?, ?> getScheme()
135 {
136 return scheme;
137 }
138
139 /**
140 * Method getTrace return the trace of this object.
141 *
142 * @return String
143 */
144 public String getTrace()
145 {
146 return trace;
147 }
148
149 /**
150 * Method flowInit allows this Tap instance to initialize itself in context of the given {@link cascading.flow.Flow} instance.
151 * This method is guaranteed to be called before the Flow is started and the
152 * {@link cascading.flow.FlowListener#onStarting(cascading.flow.Flow)} event is fired.
153 * <p/>
154 * This method will be called once per Flow, and before {@link #sourceConfInit(cascading.flow.FlowProcess, Object)} and
155 * {@link #sinkConfInit(cascading.flow.FlowProcess, Object)} methods.
156 *
157 * @param flow of type Flow
158 */
159 public void flowConfInit( Flow<Config> flow )
160 {
161
162 }
163
164 /**
165 * Method sourceConfInit initializes this instance as a source.
166 * <p/>
167 * This method maybe called more than once if this Tap instance is used outside the scope of a {@link cascading.flow.Flow}
168 * instance or if it participates in multiple times in a given Flow or across different Flows in
169 * a {@link cascading.cascade.Cascade}.
170 * <p/>
171 * In the context of a Flow, it will be called after
172 * {@link cascading.flow.FlowListener#onStarting(cascading.flow.Flow)}
173 * <p/>
174 * Note that no resources or services should be modified by this method.
175 *
176 * @param flowProcess of type FlowProcess
177 * @param conf of type Config
178 */
179 public void sourceConfInit( FlowProcess<Config> flowProcess, Config conf )
180 {
181 getScheme().sourceConfInit( flowProcess, this, conf );
182 }
183
184 /**
185 * Method sinkConfInit initializes this instance as a sink.
186 * <p/>
187 * This method maybe called more than once if this Tap instance is used outside the scope of a {@link cascading.flow.Flow}
188 * instance or if it participates in multiple times in a given Flow or across different Flows in
189 * a {@link cascading.cascade.Cascade}.
190 * <p/>
191 * Note this method will be called in context of this Tap being used as a traditional 'sink' and as a 'trap'.
192 * <p/>
193 * In the context of a Flow, it will be called after
194 * {@link cascading.flow.FlowListener#onStarting(cascading.flow.Flow)}
195 * <p/>
196 * Note that no resources or services should be modified by this method. If this Tap instance returns true for
197 * {@link #isReplace()}, then {@link #deleteResource(Object)} will be called by the parent Flow.
198 *
199 * @param flowProcess of type FlowProcess
200 * @param conf of type Config
201 */
202 public void sinkConfInit( FlowProcess<Config> flowProcess, Config conf )
203 {
204 getScheme().sinkConfInit( flowProcess, this, conf );
205 }
206
207 /**
208 * Method getIdentifier returns a String representing the resource this Tap instance represents.
209 * <p/>
210 * Often, if the tap accesses a filesystem, the identifier is nothing more than the path to the file or directory.
211 * In other cases it may be a an URL or URI representing a connection string or remote resource.
212 * <p/>
213 * Any two Tap instances having the same value for the identifier are considered equal.
214 *
215 * @return String
216 */
217 public abstract String getIdentifier();
218
219 /**
220 * Method getSourceFields returns the sourceFields of this Tap object.
221 *
222 * @return the sourceFields (type Fields) of this Tap object.
223 */
224 public Fields getSourceFields()
225 {
226 return getScheme().getSourceFields();
227 }
228
229 /**
230 * Method getSinkFields returns the sinkFields of this Tap object.
231 *
232 * @return the sinkFields (type Fields) of this Tap object.
233 */
234 public Fields getSinkFields()
235 {
236 return getScheme().getSinkFields();
237 }
238
239 /**
240 * Method openForRead opens the resource represented by this Tap instance for reading.
241 * <p/>
242 * {@code input} value may be null, if so, sub-classes must inquire with the underlying {@link Scheme}
243 * via {@link Scheme#sourceConfInit(cascading.flow.FlowProcess, Tap, Object)} to get the proper
244 * input type and instantiate it before calling {@code super.openForRead()}.
245 * <p/>
246 * Note the returned iterator will return the same instance of {@link cascading.tuple.TupleEntry} on every call,
247 * thus a copy must be made of either the TupleEntry or the underlying {@code Tuple} instance if they are to be
248 * stored in a Collection.
249 *
250 * @param flowProcess of type FlowProcess
251 * @param input of type Input
252 * @return TupleEntryIterator
253 * @throws java.io.IOException when the resource cannot be opened
254 */
255 public abstract TupleEntryIterator openForRead( FlowProcess<Config> flowProcess, Input input ) throws IOException;
256
257 /**
258 * Method openForRead opens the resource represented by this Tap instance for reading.
259 * <p/>
260 * Note the returned iterator will return the same instance of {@link cascading.tuple.TupleEntry} on every call,
261 * thus a copy must be made of either the TupleEntry or the underlying {@code Tuple} instance if they are to be
262 * stored in a Collection.
263 *
264 * @param flowProcess of type FlowProcess
265 * @return TupleEntryIterator
266 * @throws java.io.IOException when the resource cannot be opened
267 */
268 public TupleEntryIterator openForRead( FlowProcess<Config> flowProcess ) throws IOException
269 {
270 return openForRead( flowProcess, null );
271 }
272
273 /**
274 * Method openForWrite opens the resource represented by this Tap instance for writing.
275 * <p/>
276 * This method is used internally and does not honor the {@link SinkMode} setting. If SinkMode is
277 * {@link SinkMode#REPLACE}, this call may fail. See {@link #openForWrite(cascading.flow.FlowProcess)}.
278 * <p/>
279 * {@code output} value may be null, if so, sub-classes must inquire with the underlying {@link Scheme}
280 * via {@link Scheme#sinkConfInit(cascading.flow.FlowProcess, Tap, Object)} to get the proper
281 * output type and instantiate it before calling {@code super.openForWrite()}.
282 *
283 * @param flowProcess of type FlowProcess
284 * @param output of type Output
285 * @return TupleEntryCollector
286 * @throws java.io.IOException when the resource cannot be opened
287 */
288 public abstract TupleEntryCollector openForWrite( FlowProcess<Config> flowProcess, Output output ) throws IOException;
289
290 /**
291 * Method openForWrite opens the resource represented by this Tap instance for writing.
292 * <p/>
293 * This method is for user application use and does honor the {@link SinkMode#REPLACE} settings. That is, if
294 * SinkMode is set to {@link SinkMode#REPLACE} the underlying resource will be deleted.
295 * <p/>
296 * Note if {@link SinkMode#UPDATE} is set, the resource will not be deleted.
297 *
298 * @param flowProcess of type FlowProcess
299 * @return TupleEntryCollector
300 * @throws java.io.IOException when the resource cannot be opened
301 */
302 public TupleEntryCollector openForWrite( FlowProcess<Config> flowProcess ) throws IOException
303 {
304 if( isReplace() )
305 deleteResource( flowProcess.getConfigCopy() );
306
307 return openForWrite( flowProcess, null );
308 }
309
310 @Override
311 public Scope outgoingScopeFor( Set<Scope> incomingScopes )
312 {
313 // as a source Tap, we emit the scheme defined Fields
314 // as a sink Tap, we declare we emit the incoming Fields
315 // as a temp Tap, this method never gets called, but we emit what we consume
316 int count = 0;
317 for( Scope incomingScope : incomingScopes )
318 {
319 Fields incomingFields = incomingScope.getIncomingTapFields();
320
321 if( incomingFields != null )
322 {
323 try
324 {
325 incomingFields.select( getSinkFields() );
326 }
327 catch( FieldsResolverException exception )
328 {
329 throw new TapException( this, exception.getSourceFields(), exception.getSelectorFields(), exception );
330 }
331
332 count++;
333 }
334 }
335
336 if( count > 1 )
337 throw new FlowException( "Tap may not have more than one incoming Scope" );
338
339 // this allows the incoming to be passed through to the outgoing
340 Fields incomingFields = incomingScopes.size() == 0 ? null : incomingScopes.iterator().next().getIncomingTapFields();
341
342 if( incomingFields != null &&
343 ( isSource() && getSourceFields().equals( Fields.UNKNOWN ) ||
344 isSink() && getSinkFields().equals( Fields.ALL ) ) )
345 return new Scope( incomingFields );
346
347 if( count == 1 )
348 return new Scope( getSinkFields() );
349
350 return new Scope( getSourceFields() );
351 }
352
353 /**
354 * A hook for allowing a Scheme to lazily retrieve its source fields.
355 *
356 * @param flowProcess of type FlowProcess
357 * @return the found Fields
358 */
359 public Fields retrieveSourceFields( FlowProcess<Config> flowProcess )
360 {
361 return getScheme().retrieveSourceFields( flowProcess, this );
362 }
363
364 public void presentSourceFields( FlowProcess<Config> flowProcess, Fields fields )
365 {
366 getScheme().presentSourceFields( flowProcess, this, fields );
367 }
368
369 /**
370 * A hook for allowing a Scheme to lazily retrieve its sink fields.
371 *
372 * @param flowProcess of type FlowProcess
373 * @return the found Fields
374 */
375 public Fields retrieveSinkFields( FlowProcess<Config> flowProcess )
376 {
377 return getScheme().retrieveSinkFields( flowProcess, this );
378 }
379
380 public void presentSinkFields( FlowProcess<Config> flowProcess, Fields fields )
381 {
382 getScheme().presentSinkFields( flowProcess, this, fields );
383 }
384
385 @Override
386 public Fields resolveIncomingOperationArgumentFields( Scope incomingScope )
387 {
388 return incomingScope.getIncomingTapFields();
389 }
390
391 @Override
392 public Fields resolveIncomingOperationPassThroughFields( Scope incomingScope )
393 {
394 return incomingScope.getIncomingTapFields();
395 }
396
397 /**
398 * Method getFullIdentifier returns a fully qualified resource identifier.
399 *
400 * @param flowProcess of type FlowProcess
401 * @return String
402 */
403 public String getFullIdentifier( FlowProcess<Config> flowProcess )
404 {
405 return getFullIdentifier( flowProcess.getConfigCopy() );
406 }
407
408 /**
409 * Method getFullIdentifier returns a fully qualified resource identifier.
410 *
411 * @param conf of type Config
412 * @return String
413 */
414 public String getFullIdentifier( Config conf )
415 {
416 return getIdentifier();
417 }
418
419 /**
420 * Method createResource creates the underlying resource.
421 *
422 * @param flowProcess of type FlowProcess
423 * @return boolean
424 * @throws IOException when there is an error making directories
425 */
426 public boolean createResource( FlowProcess<Config> flowProcess ) throws IOException
427 {
428 return createResource( flowProcess.getConfigCopy() );
429 }
430
431 /**
432 * Method createResource creates the underlying resource.
433 *
434 * @param conf of type Config
435 * @return boolean
436 * @throws IOException when there is an error making directories
437 */
438 public abstract boolean createResource( Config conf ) throws IOException;
439
440 /**
441 * Method deleteResource deletes the resource represented by this instance.
442 *
443 * @param flowProcess of type FlowProcess
444 * @return boolean
445 * @throws IOException when the resource cannot be deleted
446 */
447 public boolean deleteResource( FlowProcess<Config> flowProcess ) throws IOException
448 {
449 return deleteResource( flowProcess.getConfigCopy() );
450 }
451
452 /**
453 * Method deleteResource deletes the resource represented by this instance.
454 *
455 * @param conf of type Config
456 * @return boolean
457 * @throws IOException when the resource cannot be deleted
458 */
459 public abstract boolean deleteResource( Config conf ) throws IOException;
460
461 /**
462 * Method commitResource allows the underlying resource to be notified when all write processing is
463 * successful so that any additional cleanup or processing may be completed.
464 * <p/>
465 * See {@link #rollbackResource(Object)} to handle cleanup in the face of failures.
466 * <p/>
467 * This method is invoked once "client side" and not in the cluster, if any.
468 * <p/>
469 * If other sink Tap instance in a given Flow fail on commitResource after called on this instance,
470 * rollbackResource will not be called.
471 * <p/>
472 * <emphasis>This is an experimental API and subject to refinement!!</emphasis>
473 *
474 * @param conf of type Config
475 * @return returns true if successful
476 * @throws IOException
477 */
478 public boolean commitResource( Config conf ) throws IOException
479 {
480 return true;
481 }
482
483 /**
484 * Method rollbackResource allows the underlying resource to be notified when any write processing has failed or
485 * was stopped so that any cleanup may be started.
486 * <p/>
487 * See {@link #commitResource(Object)} to handle cleanup when the write has successfully completed.
488 * <p/>
489 * This method is invoked once "client side" and not in the cluster, if any.
490 * <p/>
491 * <emphasis>This is an experimental API and subject to refinement!!</emphasis>
492 *
493 * @param conf of type Config
494 * @return returns true if successful
495 * @throws IOException
496 */
497 public boolean rollbackResource( Config conf ) throws IOException
498 {
499 return true;
500 }
501
502 /**
503 * Method resourceExists returns true if the path represented by this instance exists.
504 *
505 * @param flowProcess of type FlowProcess
506 * @return true if the underlying resource already exists
507 * @throws IOException when the status cannot be determined
508 */
509 public boolean resourceExists( FlowProcess<Config> flowProcess ) throws IOException
510 {
511 return resourceExists( flowProcess.getConfigCopy() );
512 }
513
514 /**
515 * Method resourceExists returns true if the path represented by this instance exists.
516 *
517 * @param conf of type Config
518 * @return true if the underlying resource already exists
519 * @throws IOException when the status cannot be determined
520 */
521 public abstract boolean resourceExists( Config conf ) throws IOException;
522
523 /**
524 * Method getModifiedTime returns the date this resource was last modified.
525 *
526 * @param flowProcess of type FlowProcess
527 * @return The date this resource was last modified.
528 * @throws IOException
529 */
530 public long getModifiedTime( FlowProcess<Config> flowProcess ) throws IOException
531 {
532 return getModifiedTime( flowProcess.getConfigCopy() );
533 }
534
535 /**
536 * Method getModifiedTime returns the date this resource was last modified.
537 *
538 * @param conf of type Config
539 * @return The date this resource was last modified.
540 * @throws IOException
541 */
542 public abstract long getModifiedTime( Config conf ) throws IOException;
543
544 /**
545 * Method getSinkMode returns the {@link SinkMode} }of this Tap object.
546 *
547 * @return the sinkMode (type SinkMode) of this Tap object.
548 */
549 public SinkMode getSinkMode()
550 {
551 return sinkMode;
552 }
553
554 /**
555 * Method isKeep indicates whether the resource represented by this instance should be kept if it
556 * already exists when the Flow is started.
557 *
558 * @return boolean
559 */
560 public boolean isKeep()
561 {
562 return sinkMode == SinkMode.KEEP;
563 }
564
565 /**
566 * Method isReplace indicates whether the resource represented by this instance should be deleted if it
567 * already exists when the Flow is started.
568 *
569 * @return boolean
570 */
571 public boolean isReplace()
572 {
573 return sinkMode == SinkMode.REPLACE;
574 }
575
576 /**
577 * Method isUpdate indicates whether the resource represented by this instance should be updated if it already
578 * exists. Otherwise a new resource will be created, via {@link #createResource(Object)}, when the Flow is started.
579 *
580 * @return boolean
581 */
582 public boolean isUpdate()
583 {
584 return sinkMode == SinkMode.UPDATE;
585 }
586
587 /**
588 * Method isSink returns true if this Tap instance can be used as a sink.
589 *
590 * @return boolean
591 */
592 public boolean isSink()
593 {
594 return getScheme().isSink();
595 }
596
597 /**
598 * Method isSource returns true if this Tap instance can be used as a source.
599 *
600 * @return boolean
601 */
602 public boolean isSource()
603 {
604 return getScheme().isSource();
605 }
606
607 /**
608 * Method isTemporary returns true if this Tap is temporary (used for intermediate results).
609 *
610 * @return the temporary (type boolean) of this Tap object.
611 */
612 public boolean isTemporary()
613 {
614 return false;
615 }
616
617 /**
618 * Returns a {@link cascading.property.ConfigDef} instance that allows for local properties to be set and made available via
619 * a resulting {@link cascading.flow.FlowProcess} instance when the tap is invoked.
620 * <p/>
621 * Any properties set on the configDef will not show up in any {@link Flow} or {@link cascading.flow.FlowStep} process
622 * level configuration, but will override any of those values as seen by the current Tap instance method call where a
623 * FlowProcess is provided except for the {@link #sourceConfInit(cascading.flow.FlowProcess, Object)} and
624 * {@link #sinkConfInit(cascading.flow.FlowProcess, Object)} methods.
625 * <p/>
626 * That is, the {@code *confInit} methods are called before any ConfigDef is applied, so any values placed into
627 * a ConfigDef instance will not be visible to them.
628 *
629 * @return an instance of ConfigDef
630 */
631 public ConfigDef getConfigDef()
632 {
633 if( configDef == null )
634 configDef = new ConfigDef();
635
636 return configDef;
637 }
638
639 /**
640 * Returns {@code true} if there are properties in the configDef instance.
641 *
642 * @return true if there are configDef properties
643 */
644 public boolean hasConfigDef()
645 {
646 return configDef != null && !configDef.isEmpty();
647 }
648
649 /**
650 * Returns a {@link ConfigDef} instance that allows for process level properties to be set and made available via
651 * a resulting {@link cascading.flow.FlowProcess} instance when the tap is invoked.
652 * <p/>
653 * Any properties set on the stepConfigDef will not show up in any Flow configuration, but will show up in
654 * the current process {@link cascading.flow.FlowStep} (in Hadoop the MapReduce jobconf). Any value set in the
655 * stepConfigDef will be overridden by the tap local {@code #getConfigDef} instance.
656 * </p>
657 * Use this method to tweak properties in the process step this tap instance is planned into.
658 * <p/>
659 * Note the {@code *confInit} methods are called before any ConfigDef is applied, so any values placed into
660 * a ConfigDef instance will not be visible to them.
661 *
662 * @return an instance of ConfigDef
663 */
664 @Override
665 public ConfigDef getStepConfigDef()
666 {
667 if( processConfigDef == null )
668 processConfigDef = new ConfigDef();
669
670 return processConfigDef;
671 }
672
673 /**
674 * Returns {@code true} if there are properties in the processConfigDef instance.
675 *
676 * @return true if there are processConfigDef properties
677 */
678 @Override
679 public boolean hasStepConfigDef()
680 {
681 return processConfigDef != null && !processConfigDef.isEmpty();
682 }
683
684 @Override
685 public boolean isEquivalentTo( FlowElement element )
686 {
687 if( element == null )
688 return false;
689
690 if( this == element )
691 return true;
692
693 boolean compare = getClass() == element.getClass();
694
695 if( !compare )
696 return false;
697
698 return equals( element );
699 }
700
701 @Override
702 public boolean equals( Object object )
703 {
704 if( this == object )
705 return true;
706 if( object == null || getClass() != object.getClass() )
707 return false;
708
709 Tap tap = (Tap) object;
710
711 if( getIdentifier() != null ? !getIdentifier().equals( tap.getIdentifier() ) : tap.getIdentifier() != null )
712 return false;
713
714 if( getScheme() != null ? !getScheme().equals( tap.getScheme() ) : tap.getScheme() != null )
715 return false;
716
717 return true;
718 }
719
720 @Override
721 public int hashCode()
722 {
723 int result = getIdentifier() != null ? getIdentifier().hashCode() : 0;
724
725 result = 31 * result + ( getScheme() != null ? getScheme().hashCode() : 0 );
726
727 return result;
728 }
729
730 @Override
731 public String toString()
732 {
733 if( getIdentifier() != null )
734 return getClass().getSimpleName() + "[\"" + getScheme() + "\"]" + "[\"" + Util.sanitizeUrl( getIdentifier() ) + "\"]"; // sanitize
735 else
736 return getClass().getSimpleName() + "[\"" + getScheme() + "\"]" + "[not initialized]";
737 }
738 }