Hi Urvasi,

1) Bulk api always returns csv files as a result.
2) function getBulkConnection creates connection with salesforce, you need to provide your
    user name and password (password + token).
3) createJob function creates job which will execute bulk operation.
4) createBatchesFromCSVFile function creates result csv file and saves to path given in "csvFileName" parameter.

Below is the example which I used in my requirement.

//This is first function which I calls when I need to execute my query using bulk api.

public void retrieveBulkData(String userName, String password,
		String objectName, String query, String fileName, String environment)
		throws ConnectionException, AsyncApiException,
		FileNotFoundException, IOException {
		
	long start=System.currentTimeMillis();	
	BulkConnection bulkconnection = getBulkConnection(userName,password,environment);
	JobInfo job = createJob(objectName,bulkconnection);
	//String query = "SELECT Name,Id FROM contact";
	ByteArrayInputStream byteArrayInputStream =getData(query,bulkconnection,job);
	closeJob(bulkconnection, job.getId());
	System.out.println("FileName::"+fileName);
	createCSVFile(byteArrayInputStream, fileName);
	long end=System.currentTimeMillis();
	
	long total=start-end;
	System.out.println("total:"+total);
	int seconds=(int) ((total/1000)%60);
	int minutes=(int) ((total / (1000 * 60)) % 60);
	System.out.println("Time To Execute is::"+minutes+":"+seconds);
}

 //Below function I am using to create connection with salesforce

public BulkConnection getBulkConnection(String userName, String password,
		String environment) throws ConnectionException, AsyncApiException {
	BulkConnection bulkConnection = null;
	try {
		ConnectorConfig config = new ConnectorConfig();
		config.setUsername(userName);
		config.setPassword(password);
		System.out.println("BulkAPI.java file environment: "+environment);
		if(environment.trim().equalsIgnoreCase("Production")){
			System.out.println("conn for production type");
			config.setAuthEndpoint("https://login.salesforce.com/services/Soap/u/26.0");
		}
		else if(environment.trim().equalsIgnoreCase("Sandbox")){
			System.out.println("conn for sand box type");
			config.setAuthEndpoint("https://test.salesforce.com/services/Soap/u/26.0");
		}
		else{
			System.out.println("conn else case");
			config.setAuthEndpoint("https://login.salesforce.com/services/Soap/u/26.0");
		}
		config.setCompression(true);
		config.setTraceFile("traceLogs.txt");
		config.setTraceMessage(true);
		config.setPrettyPrintXml(true);
		System.out.println("AuthEndpoint: " + config.getRestEndpoint());
		new PartnerConnection(config);
		System.out.println("SessionID: " + config.getSessionId());
		// String soapAuthEndPoint = config.getServiceEndpoint();;
		// config.setRestEndpoint(bulkAuthEndPoint);
		String soapEndpoint = config.getServiceEndpoint();
		String apiVersion = "26.0";
		String restEndpoint = soapEndpoint.substring(0,
				soapEndpoint.indexOf("Soap/"))
				+ "async/" + apiVersion;
		config.setRestEndpoint(restEndpoint);
		// This should only be false when doing debugging.
		config.setCompression(true);
		bulkConnection = new BulkConnection(config);
		
	} 
	
	catch (AsyncApiException aae) {
		aae.printStackTrace();
	} catch (ConnectionException ce) {
		ce.printStackTrace();
	} catch (FileNotFoundException fnfe) {
		fnfe.printStackTrace();
	}
	return bulkConnection;
	
}

 
//Below function creates jobs

public JobInfo createJob(String sobjectType, BulkConnection bulkconnection)
		throws AsyncApiException {
	JobInfo job = new JobInfo();
	job.setObject(sobjectType);
	job.setOperation(OperationEnum.query);
	job.setConcurrencyMode(ConcurrencyMode.Parallel);
	job.setContentType(ContentType.CSV);
	job = bulkconnection.createJob(job);
	assert job.getId() != null;
	job = bulkconnection.getJobStatus(job.getId());
	System.out.println(job);
	return job;
}

 //Below function executes query to fetch data.

public ByteArrayInputStream getData(String query,
		BulkConnection bulkConnection, JobInfo job) {
	ByteArrayInputStream inputStream = null;
	try{
		BatchInfo info = null;
		ByteArrayInputStream bout = new ByteArrayInputStream(
				query.getBytes());// convert query into ByteArrayInputStream
		info = bulkConnection.createBatchFromStream(job, bout);// creates batch from ByteArrayInputStream
		String[] queryResults = null;// to store QueryResultList
		QueryResultList list = null;
		int count=0;
		for (int i = 0; i < 10000; i++) // 10000=maxRowsPerBatch
		{
			count++;
			Thread.sleep(i == 0 ? 30 * 1000 : 30 * 1000); // 30 sec
			
			info = bulkConnection.getBatchInfo(job.getId(), info.getId());
			// If Batch Status is Completed,get QueryResultList and store in
			// queryResults.
			if (info.getState() == BatchStateEnum.Completed) {
				list = bulkConnection.getQueryResultList(job.getId(),
						info.getId());
				queryResults = list.getResult();

				break;
			} else if (info.getState() == BatchStateEnum.Failed) {
				System.out.println("-------------- failed ----------"
						+ info);
				break;
			} else {
				System.out.println("-------------- waiting ----------"
						+ info);
			}
		}
		System.out.println("count::--"+count);
		System.out.println("QueryResultList::" + list.toString());
		if (queryResults != null) {
			//for each resultid retrieve data and store in ByteArrayInputstream provided job id,batch id and result id 
			//as returnType for bulkConnection.getQueryResultStream() method is ByteArrayInputstream
			for (String resultId : queryResults) {
				inputStream=(ByteArrayInputStream) bulkConnection.getQueryResultStream(job.getId(),
						info.getId(), resultId);
			}
		}
		
	} catch (AsyncApiException aae) {
		aae.printStackTrace();
	} catch (InterruptedException ie) {
		ie.printStackTrace();
	}
	return inputStream;
}

 //I am using following function when i need to do insert/update operation to salesforce from csv file

private List<BatchInfo> createBatchesFromCSVFile(BulkConnection connection,JobInfo jobInfo, String csvFileName) throws IOException,	AsyncApiException {
	List<BatchInfo> batchInfos = new ArrayList<BatchInfo>();
	BufferedReader rdr = new BufferedReader(new InputStreamReader(
			new FileInputStream(csvFileName)));
	// read the CSV header row
	CommonPropertyManager cpm = PropertyManagerFactory.getCommonPropertyManager();
	String dirName=cpm.getPath().getTemporaryFilesPath() + java.io.File.separator +"salesforce.com"+ java.io.File.separator;
	byte[] headerBytes = (rdr.readLine() + "\n").getBytes("UTF-8");
	int headerBytesLength = headerBytes.length;
	//File tmpFile = File.createTempFile("bulkAPIInsert", ".csv");
	File tmpFile = File.createTempFile("bulkAPIUpdate", ".csv");
	// Split the CSV file into multiple batches
	try {
		FileOutputStream tmpOut = new FileOutputStream(tmpFile);
		int maxBytesPerBatch = 10000000; // 10 million bytes per batch
		int maxRowsPerBatch = 10000; // 10 thousand rows per batch
		int currentBytes = 0;
		int currentLines = 0;
		String nextLine;
		while ((nextLine = rdr.readLine()) != null) {
			System.out.println("In while");
			byte[] bytes = (nextLine + "\n").getBytes("UTF-8");
			// Create a new batch when our batch size limit is reached
			if (currentBytes + bytes.length > maxBytesPerBatch
					|| currentLines > maxRowsPerBatch) {
				createBatch(tmpOut, tmpFile, batchInfos, connection,
						jobInfo);
				currentBytes = 0;
				currentLines = 0;
			}
			if (currentBytes == 0) {
				tmpOut = new FileOutputStream(tmpFile);
				tmpOut.write(headerBytes);
				currentBytes = headerBytesLength;
				currentLines = 1;
			}
			tmpOut.write(bytes);
			currentBytes += bytes.length;
			currentLines++;
		}
		// Finished processing all rows
		// Create a final batch for any remaining data
		System.out.println("After While");
		if (currentLines > 1) {
			System.out.println("before createBatch");
			createBatch(tmpOut, tmpFile, batchInfos, connection, jobInfo);
			System.out.println("before createBatch");
		}
	} finally {
		rdr.close();
		try
		{
			tmpFile.delete();	
		}
		catch (Exception e)
		{
			System.out.println("Exception in deleting file");
		}
		
	}
	return batchInfos;
}

private void createBatch(FileOutputStream tmpOut, File tmpFile,List<BatchInfo> batchInfos, BulkConnection connection,JobInfo jobInfo) throws IOException, AsyncApiException {
    tmpOut.flush();
    tmpOut.close();
    FileInputStream tmpInputStream = new FileInputStream(tmpFile);
    try {
        BatchInfo batchInfo = connection.createBatchFromStream(jobInfo,
                tmpInputStream);
        System.out.println(batchInfo);
        batchInfos.add(batchInfo);
    }
    catch(Exception ex){
        System.out.println("##### Exception in createbatch " + ex);
    } finally {
        tmpInputStream.close();
    }
}

 I hope this will be helpful to you.